08_pipe.c

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
// Example 08 -- two fibers talking over a pipe with async IO.
//
// srn_fiber_read / srn_fiber_write look like blocking read/write, but they
// suspend the calling fiber and hand the operation to the reactor rather than
// blocking the worker os thread. The worker runs the other fiber meanwhile, and
// the reactor wakes the issuer when its transfer completes. The descriptors must
// be non-blocking so the reactor can drive them on readiness.
//
// A writer fiber sends a line into the pipe and a reader fiber pulls it out. On
// a single worker whichever runs first parks in its IO call and yields to the
// other, so the transfer completes cooperatively.
 
#include <fcntl.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
 
#include <serene/rt/fiber/io.h>
#include <serene/runtime.h>
 
typedef struct {
  int fd;
  const char *msg;
} writer_args_t;
 
typedef struct {
  int fd;
  char *buf;
  size_t cap;
} reader_args_t;
 
static void set_nonblock(int fd) {
  int fl = fcntl(fd, F_GETFL, 0);
  (void)fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}
 
static srn_fiber_result_t writer(srn_context_t *ctx, void *arg) {
  (void)ctx;
  writer_args_t *w       = arg;
  srn_io_size_result_t r = srn_fiber_write(w->fd, w->msg, strlen(w->msg), -1);
  if (HAS_ERROR(r)) {
    printf("writer: error: %s\n", r.maybe_error->msg);
  } else {
    printf("writer: wrote %zu bytes\n", r.count);
  }
  close(w->fd); // signals EOF to the reader once the bytes are drained
  return nullptr;
}
 
static srn_fiber_result_t reader(srn_context_t *ctx, void *arg) {
  (void)ctx;
  reader_args_t *rd      = arg;
  srn_io_size_result_t r = srn_fiber_read(rd->fd, rd->buf, rd->cap - 1, -1);
  if (HAS_ERROR(r)) {
    printf("reader: error: %s\n", r.maybe_error->msg);
  } else {
    rd->buf[r.count] = '\0';
    printf("reader: read %zu bytes: \"%s\"\n", r.count, rd->buf);
  }
  close(rd->fd);
  return nullptr;
}
 
int main(void) {
  SERENE_RUNTIME_INIT_WITH_SCHED(engine, sched, nullptr);
  srn_context_t *ctx = srn_context_make(engine);
 
  int fds[2];
  if (pipe(fds) != 0) {
    perror("pipe");
    return 1;
  }
  set_nonblock(fds[0]);
  set_nonblock(fds[1]);
 
  char buf[64]    = {0};
  writer_args_t w = {fds[1], "hello over a pipe"};
  reader_args_t r = {fds[0], buf, sizeof(buf)};
  (void)srn_fiber_spawn(ctx, writer, &w);
  (void)srn_fiber_spawn(ctx, reader, &r);
 
  srn_sched_run(sched, 1);
 
  srn_context_release(ctx);
  SERENE_RUNTIME_SHUTDOWN(engine);
  return 0;
}