silo/masstree/kvio.cc

   1 /* Masstree
   2  * Eddie Kohler, Yandong Mao, Robert Morris
   3  * Copyright (c) 2012-2014 President and Fellows of Harvard College
   4  * Copyright (c) 2012-2014 Massachusetts Institute of Technology
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a
   7  * copy of this software and associated documentation files (the "Software"),
   8  * to deal in the Software without restriction, subject to the conditions
   9  * listed in the Masstree LICENSE file. These conditions include: you must
  10  * preserve this copyright notice, and you cannot mention the copyright
  11  * holders in advertising related to the Software without their permission.
  12  * The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
  13  * notice is a summary of the Masstree LICENSE file; the license in that file
  14  * is legally binding.
  15  */
  16 // buffered read and write for kvc/kvd.
  17 // stdio is good but not quite what I want.
  18 // need to be able to check if any input
  19 // available, and do non-blocking check.
  20 // also, fwrite just isn't very fast, at
  21 // least on the Mac.
  22
  23 #include <stdlib.h>
  24 #include <unistd.h>
  25 #include <string.h>
  26 #include <sys/select.h>
  27 #include <sys/time.h>
  28 #include <assert.h>
  29 #include <stdio.h>
  30 #include <errno.h>
  31 #include "kvio.hh"
  32
  33
  34 // API to allocate a new kvout.
  35 kvout* new_kvout(int fd, int buflen) {
  36     kvout* kv = (kvout*) malloc(sizeof(kvout));
  37     assert(kv);
  38     memset(kv, 0, sizeof(*kv));
  39     kv->capacity = buflen;
  40     kv->buf = (char*) malloc(kv->capacity);
  41     assert(kv->buf);
  42     kv->fd = fd;
  43     return kv;
  44 }
  45
  46 // API to allocate a new kvout for a buffer, no fd.
  47 kvout* new_bufkvout() {
  48     kvout *kv = (kvout*) malloc(sizeof(kvout));
  49     assert(kv);
  50     memset(kv, 0, sizeof(*kv));
  51     kv->capacity = 256;
  52     kv->buf = (char*) malloc(kv->capacity);
  53     assert(kv->buf);
  54     kv->n = 0;
  55     kv->fd = -1;
  56     return kv;
  57 }
  58
  59 // API to clear out a buf kvout.
  60 void kvout_reset(kvout* kv) {
  61     assert(kv->fd < 0);
  62     kv->n = 0;
  63 }
  64
  65 // API to free a kvout.
  66 // does not close() the fd.
  67 void free_kvout(kvout* kv) {
  68     if (kv->buf)
  69         free(kv->buf);
  70     kv->buf = 0;
  71     free(kv);
  72 }
  73
  74 void kvflush(kvout* kv) {
  75     assert(kv->fd >= 0);
  76     size_t sent = 0;
  77     while (kv->n > sent) {
  78         ssize_t cc = write(kv->fd, kv->buf + sent, kv->n - sent);
  79         if (cc <= 0) {
  80             if (errno == EWOULDBLOCK) {
  81                 usleep(1);
  82                 continue;
  83             }
  84             perror("kvflush write");
  85             return;
  86         }
  87         sent += cc;
  88     }
  89     kv->n = 0;
  90 }
  91
  92 // API
  93 void kvout::grow(unsigned want) {
  94     if (fd >= 0)
  95         kvflush(this);
  96     if (want == 0)
  97         want = capacity + 1;
  98     while (want > capacity)
  99         capacity *= 2;
 100     buf = (char*) realloc(buf, capacity);
 101     assert(buf);
 102 }
 103
 104 int kvwrite(kvout* kv, const void* buf, unsigned n) {
 105     if (kv->n + n > kv->capacity && kv->fd >= 0)
 106         kvflush(kv);
 107     if (kv->n + n > kv->capacity)
 108         kv->grow(kv->n + n);
 109     memcpy(kv->buf + kv->n, buf, n);
 110     kv->n += n;
 111     return n;
 112 }