SchOS/kernel/net/unix.c

// Unix domain sockets.
//
// A connected pair is two buffers, not one. Writing to a socket appends to
// its *peer's* buffer and reading drains its own, which is what makes a
// socket bidirectional where a pipe is not, without needing two of anything.
//
// **One lock covers the whole table**, rather than a lock per socket. A pipe
// gets away with a lock per pipe because every operation touches exactly one
// object; a socket operation touches two, and two-object locking needs an
// ordering rule that every path obeys or it deadlocks. Worse, the wakeup
// races are subtle: a reader that tests `peer_closed` under its own lock while
// the closer sets it under the peer's has no interlock at all, and sleeps
// through the wakeup that was meant for it. With one lock the condition and
// the sleep are atomic with respect to every writer of that condition, which
// is the property sleep/wakeup actually requires. Thirty-two sockets carrying
// GUI messages will not notice the contention.
//
// **Names are held in a table here, not in the filesystem.** A real Unix
// socket appears in a directory listing as a special file, and binding one
// creates that node. Doing that properly means a new inode type threaded
// through both filesystems, iget, dirlookup and stat -- a VFS change, not a
// socket change. The table gives every behaviour that matters: a client
// finds a server by a name it agreed in advance, which is the whole job.
// What is lost is `ls -l /tmp/schcom` showing an `s`, and `unlink()` of a
// socket path failing with ENOENT. Both are worth fixing when the VFS next
// gets opened up; neither blocks anything.

#include "types.h"
#include "defs.h"
#include "param.h"
#include "mmu.h"
#include "proc.h"
#include "fs.h"
#include "spinlock.h"
#include "sleeplock.h"
#include "file.h"
#include "posix.h"
#include "errno.h"

#define _KERNEL_SOCKET_INTERNALS
#include "socket.h"

static struct {
  struct spinlock lock;
  struct socket sock[NSOCK];
} stab;

void
sockinit(void)
{
  initlock(&stab.lock, "socktab");
}

// Caller holds stab.lock.
static struct socket *
alloc_locked(int type)
{
  struct socket *s;

  for(s = stab.sock; s < stab.sock + NSOCK; s++){
    if(s->state == SS_FREE && s->ref == 0){
      memset(s, 0, sizeof(*s));
      s->ref   = 1;
      s->type  = type;
      s->state = SS_UNBOUND;
      return s;
    }
  }
  return 0;
}

// Find a socket listening on this name. Caller holds stab.lock.
static struct socket *
find_listener(const char *name)
{
  struct socket *s;

  for(s = stab.sock; s < stab.sock + NSOCK; s++)
    if(s->state == SS_LISTENING && strncmp(s->name, name, UNIX_PATH_MAX) == 0)
      return s;
  return 0;
}

// Detach s from its peer and tell the peer about it. Caller holds stab.lock.
static void
orphan_peer(struct socket *s)
{
  struct socket *peer = s->peer;

  if(peer == 0)
    return;
  peer->peer_closed = 1;
  peer->peer = 0;
  s->peer = 0;

  // Both channels, because the peer may be blocked reading (waiting for data
  // that will now never come) or writing (waiting for room in a buffer nobody
  // will ever drain).
  wakeup(&peer->nread);
  wakeup(&peer->nwrite);
}

void
sockclose(struct socket *s)
{
  int i;

  if(s == 0)
    return;

  acquire(&stab.lock);
  if(--s->ref > 0){
    release(&stab.lock);
    return;
  }

  orphan_peer(s);

  // A listener that goes away takes its unaccepted connections with it. Each
  // queued entry is the *server* side of a connection whose client is already
  // live and may be writing into it; that client has to be told, or it hangs
  // on a socket that will never be read.
  for(i = 0; i < s->nqueue; i++){
    struct socket *q = s->queue[i];
    if(q == 0)
      continue;
    orphan_peer(q);
    q->state = SS_FREE;
    q->ref   = 0;
  }
  s->nqueue = 0;

  s->state = SS_FREE;
  s->ref   = 0;
  wakeup(&s->nqueue);           // an accept() blocked on this listener
  release(&stab.lock);
}

void
sockdup(struct socket *s)
{
  acquire(&stab.lock);
  s->ref++;
  release(&stab.lock);
}

const char *
sockname(struct socket *s)
{
  return s->name;
}

// ---------------------------------------------------------------------------
// Data
// ---------------------------------------------------------------------------

int
sockread(struct socket *s, char *addr, int n)
{
  int i;

  if(n < 0)
    return -EINVAL;

  acquire(&stab.lock);
  while(s->nread == s->nwrite && !s->peer_closed){
    if(myproc()->killed || s->state == SS_FREE){
      release(&stab.lock);
      return -EINTR;
    }
    sleep(&s->nread, &stab.lock);
  }

  for(i = 0; i < n; i++){
    if(s->nread == s->nwrite)
      break;
    addr[i] = s->data[s->nread++ % SOCKBUF];
  }

  // Waking the writer matters even when nothing was read: it may be blocked
  // on a full buffer that this call has just made room in.
  wakeup(&s->nwrite);
  release(&stab.lock);
  return i;
}

int
sockwrite(struct socket *s, char *addr, int n)
{
  struct socket *dst;
  int i;

  if(n < 0)
    return -EINVAL;

  acquire(&stab.lock);

  // Data goes into the peer's buffer, because that is where the peer will
  // look for it. Same contract as a pipe with no reader: a signal, then an
  // error. A process that writes to a dead connection and is not told about
  // it spins forever believing it is making progress.
  if(s->peer == 0){
    release(&stab.lock);
    sigsend(myproc()->pid, SIGPIPE);
    return -EPIPE;
  }

  for(i = 0; i < n; i++){
    // Re-read every iteration: the peer can go away while we sleep below.
    dst = s->peer;
    while(dst && dst->nwrite == dst->nread + SOCKBUF){
      if(myproc()->killed){
        release(&stab.lock);
        return i > 0 ? i : -EINTR;
      }
      wakeup(&dst->nread);
      sleep(&dst->nwrite, &stab.lock);
      dst = s->peer;
    }
    if(dst == 0){
      release(&stab.lock);
      sigsend(myproc()->pid, SIGPIPE);
      return i > 0 ? i : -EPIPE;    // a short write is still a write
    }
    dst->data[dst->nwrite++ % SOCKBUF] = addr[i];
  }

  wakeup(&s->peer->nread);
  release(&stab.lock);
  return i;
}

// ---------------------------------------------------------------------------
// Connection setup
// ---------------------------------------------------------------------------

struct socket *
socknew(int domain, int type)
{
  struct socket *s;

  if(domain != AF_UNIX)
    return 0;
  if(type != SOCK_STREAM)
    return 0;

  acquire(&stab.lock);
  s = alloc_locked(type);
  release(&stab.lock);
  return s;
}

// A connected pair with no name and no listener, for socketpair(2).
int
sockpair(struct socket **pa, struct socket **pb)
{
  struct socket *a, *b;

  acquire(&stab.lock);
  if((a = alloc_locked(SOCK_STREAM)) == 0){
    release(&stab.lock);
    return -ENFILE;
  }
  if((b = alloc_locked(SOCK_STREAM)) == 0){
    a->state = SS_FREE;
    a->ref = 0;
    release(&stab.lock);
    return -ENFILE;
  }
  a->peer = b; a->state = SS_CONNECTED;
  b->peer = a; b->state = SS_CONNECTED;
  release(&stab.lock);

  *pa = a;
  *pb = b;
  return 0;
}

int
sockbind(struct socket *s, const char *path)
{
  struct socket *other;

  if(path[0] == '\0')
    return -EINVAL;

  acquire(&stab.lock);
  if(s->state != SS_UNBOUND){
    release(&stab.lock);
    return -EINVAL;
  }
  for(other = stab.sock; other < stab.sock + NSOCK; other++){
    if(other != s && other->state >= SS_BOUND &&
       strncmp(other->name, path, UNIX_PATH_MAX) == 0){
      release(&stab.lock);
      return -EADDRINUSE;
    }
  }
  safestrcpy(s->name, path, UNIX_PATH_MAX);
  s->state = SS_BOUND;
  release(&stab.lock);
  return 0;
}

int
socklisten(struct socket *s, int backlog)
{
  (void)backlog;                // the queue is a fixed NBACKLOG either way

  acquire(&stab.lock);
  if(s->state != SS_BOUND && s->state != SS_LISTENING){
    release(&stab.lock);
    return -EINVAL;
  }
  s->state = SS_LISTENING;
  release(&stab.lock);
  return 0;
}

// Returns the server-side socket of a new connection with a reference already
// held, or 0 on error.
struct socket *
sockaccept(struct socket *s)
{
  struct socket *c;
  int i;

  acquire(&stab.lock);
  if(s->state != SS_LISTENING){
    release(&stab.lock);
    return 0;
  }

  while(s->nqueue == 0){
    if(myproc()->killed || s->state != SS_LISTENING){
      release(&stab.lock);
      return 0;
    }
    sleep(&s->nqueue, &stab.lock);
  }

  c = s->queue[0];
  for(i = 0; i + 1 < s->nqueue; i++)
    s->queue[i] = s->queue[i+1];
  s->nqueue--;
  release(&stab.lock);

  return c;
}

int
sockconnect(struct socket *s, const char *path)
{
  struct socket *srv, *peer;

  acquire(&stab.lock);
  if(s->state != SS_UNBOUND && s->state != SS_BOUND){
    release(&stab.lock);
    return -EINVAL;
  }
  srv = find_listener(path);
  if(srv == 0){
    release(&stab.lock);
    return -ECONNREFUSED;
  }
  if(srv->nqueue >= NBACKLOG){
    release(&stab.lock);
    return -EAGAIN;
  }

  // Every connection needs a socket at each end. The listener is not one of
  // them -- it stays listening, and accept() hands back this new one.
  peer = alloc_locked(s->type);
  if(peer == 0){
    release(&stab.lock);
    return -ENFILE;
  }

  peer->peer  = s;
  peer->state = SS_CONNECTED;
  s->peer     = peer;
  s->state    = SS_CONNECTED;

  srv->queue[srv->nqueue++] = peer;
  wakeup(&srv->nqueue);         // an accept() may be blocked on this

  release(&stab.lock);

  // Deliberately does *not* wait for accept(). On Unix a queued connection is
  // already established: the client may write, and the bytes sit in the
  // server-side buffer until somebody accepts and reads them. Blocking here
  // instead would deadlock any server that connects to itself, and would make
  // a busy server's backlog useless.
  return 0;
}

// shutdown(2). There is one buffer per direction and no per-direction flag,
// so a half-close is approximated: SHUT_RD stops this end reading anything
// further, and SHUT_WR/SHUT_RDWR give the peer end-of-file. What programs
// actually use shutdown for -- "I have finished sending, read until EOF and
// then reply" -- works; what is missing is the ability to keep reading after
// SHUT_WR, since the connection is torn down rather than half-closed.
int
sock_shutdown(struct socket *s, int how)
{
  acquire(&stab.lock);
  if(s->state != SS_CONNECTED){
    release(&stab.lock);
    return -ENOTCONN;
  }

  if(how == SHUT_RD || how == SHUT_RDWR){
    s->peer_closed = 1;
    wakeup(&s->nread);
  }
  if(how == SHUT_WR || how == SHUT_RDWR)
    orphan_peer(s);
  release(&stab.lock);
  return 0;
}

// ---------------------------------------------------------------------------
// Readiness, for poll/select
// ---------------------------------------------------------------------------

int
sockreadable(struct socket *s)
{
  if(s->state == SS_LISTENING)
    return s->nqueue > 0;
  return s->nread != s->nwrite || s->peer_closed;
}

// The peer is gone and everything it sent has been read.
int
sock_hup(struct socket *s)
{
  if(s->state == SS_LISTENING)
    return 0;
  return s->peer == 0 && s->nread == s->nwrite;
}