libatalk/dsi/dsi_stream.c

   1 /*
   2  * Copyright (c) 1998 Adrian Sun (asun@zoology.washington.edu)
   3  * All rights reserved. See COPYRIGHT.
   4  *
   5  * this file provides the following functions:
   6  * dsi_stream_write:    just write a bunch of bytes.
   7  * dsi_stream_read:     just read a bunch of bytes.
   8  * dsi_stream_send:     send a DSI header + data.
   9  * dsi_stream_receive:  read a DSI header + data.
  10  */
  11
  12 #ifdef HAVE_CONFIG_H
  13 #include "config.h"
  14 #endif /* HAVE_CONFIG_H */
  15
  16 #include <stdio.h>
  17 #include <stdlib.h>
  18
  19 #ifdef HAVE_UNISTD_H
  20 #include <unistd.h>
  21 #endif
  22
  23 #include <string.h>
  24 #include <errno.h>
  25 #include <sys/types.h>
  26 #include <sys/socket.h>
  27 #include <sys/uio.h>
  28
  29 #include <atalk/logger.h>
  30 #include <atalk/dsi.h>
  31 #include <netatalk/endian.h>
  32 #include <atalk/util.h>
  33
  34 #define min(a,b)  ((a) < (b) ? (a) : (b))
  35
  36 #ifndef MSG_MORE
  37 #define MSG_MORE 0x8000
  38 #endif
  39
  40 #ifndef MSG_DONTWAIT
  41 #define MSG_DONTWAIT 0x40
  42 #endif
  43
  44 /* ----------------------
  45    afpd is sleeping too much while trying to send something.
  46    May be there's no reader or the reader is also sleeping in write,
  47    look if there's some data for us to read, hopefully it will wake up
  48    the reader so we can write again.
  49 */
  50 static int dsi_peek(DSI *dsi)
  51 {
  52     fd_set readfds, writefds;
  53     int    len;
  54     int    maxfd;
  55     int    ret;
  56
  57     LOG(log_debug, logtype_dsi, "dsi_peek");
  58
  59     FD_ZERO(&readfds);
  60     FD_ZERO(&writefds);
  61     FD_SET( dsi->socket, &readfds);
  62     FD_SET( dsi->socket, &writefds);
  63     maxfd = dsi->socket +1;
  64
  65     while (1) {
  66         FD_SET( dsi->socket, &readfds);
  67         FD_SET( dsi->socket, &writefds);
  68
  69         /* No timeout: if there's nothing to read nor nothing to write,
  70          * we've got nothing to do at all */
  71         if ((ret = select( maxfd, &readfds, &writefds, NULL, NULL)) <= 0) {
  72             if (ret == -1 && errno == EINTR)
  73                 /* we might have been interrupted by out timer, so restart select */
  74                 continue;
  75             /* give up */
  76             LOG(log_error, logtype_dsi, "dsi_peek: unexpected select return: %d %s",
  77                 ret, ret < 0 ? strerror(errno) : "");
  78             return -1;
  79         }
  80
  81         /* Check if there's sth to read, hopefully reading that will unblock the client */
  82         if (FD_ISSET(dsi->socket, &readfds)) {
  83             len = dsi->end - dsi->eof;
  84             LOG(log_note, logtype_dsi, "dsi_peek: used read buffer space: %d bytes", dsi->eof - dsi->buffer);
  85
  86             if (len <= 0) {
  87                 /* ouch, our buffer is full ! fall back to blocking IO
  88                  * could block and disconnect but it's better than a cpu hog */
  89                 LOG(log_warning, logtype_dsi, "dsi_peek: read buffer is full");
  90                 break;
  91             }
  92
  93             if ((len = read(dsi->socket, dsi->eof, len)) <= 0) {
  94                 if (len == 0) {
  95                     LOG(log_error, logtype_dsi, "dsi_peek: EOF");
  96                     return -1;
  97                 }
  98                 LOG(log_error, logtype_dsi, "dsi_peek: read: %s", strerror(errno));
  99                 if (errno == EAGAIN)
 100                     continue;
 101                 return -1;
 102             }
 103             LOG(log_debug, logtype_dsi, "dsi_peek: read %d bytes", len);
 104
 105             dsi->eof += len;
 106         }
 107
 108         if (FD_ISSET(dsi->socket, &writefds)) {
 109             /* we can write again */
 110             LOG(log_debug, logtype_dsi, "dsi_peek: can write again");
 111             break;
 112         }
 113     }
 114
 115     return 0;
 116 }
 117
 118 /* ------------------------------
 119  * write raw data. return actual bytes read. checks against EINTR
 120  * aren't necessary if all of the signals have SA_RESTART
 121  * specified. */
 122 ssize_t dsi_stream_write(DSI *dsi, void *data, const size_t length, int mode)
 123 {
 124   size_t written;
 125   ssize_t len;
 126   unsigned int flags = 0;
 127
 128   dsi->in_write++;
 129   written = 0;
 130
 131   LOG(log_maxdebug, logtype_dsi, "dsi_stream_write: sending %u bytes", length);
 132
 133   while (written < length) {
 134       len = send(dsi->socket, (u_int8_t *) data + written, length - written, flags);
 135       if (len >= 0) {
 136           written += len;
 137           continue;
 138       }
 139
 140       if (errno == EINTR)
 141           continue;
 142
 143       if (errno == EAGAIN || errno == EWOULDBLOCK) {
 144           LOG(log_debug, logtype_dsi, "dsi_stream_write: send: %s", strerror(errno));
 145
 146           if (mode == DSI_NOWAIT && written == 0) {
 147               /* DSI_NOWAIT is used by attention give up in this case. */
 148               written = -1;
 149               goto exit;
 150           }
 151
 152           /* Try to read sth. in order to break up possible deadlock */
 153           if (dsi_peek(dsi) != 0) {
 154               written = -1;
 155               goto exit;
 156           }
 157           /* Now try writing again */
 158           continue;
 159       }
 160
 161       LOG(log_error, logtype_dsi, "dsi_stream_write: %s", strerror(errno));
 162       written = -1;
 163       goto exit;
 164   }
 165
 166   dsi->write_count += written;
 167
 168 exit:
 169   dsi->in_write--;
 170   return written;
 171 }
 172
 173
 174 /* ---------------------------------
 175 */
 176 #ifdef WITH_SENDFILE
 177 ssize_t dsi_stream_read_file(DSI *dsi, int fromfd, off_t offset, const size_t length)
 178 {
 179   size_t written;
 180   ssize_t len;
 181
 182   LOG(log_maxdebug, logtype_dsi, "dsi_stream_read_file: sending %u bytes", length);
 183
 184   dsi->in_write++;
 185   written = 0;
 186
 187   while (written < length) {
 188     len = sys_sendfile(dsi->socket, fromfd, &offset, length - written);
 189
 190     if (len < 0) {
 191       if (errno == EINTR)
 192           continue;
 193       if (errno == EINVAL || errno == ENOSYS)
 194           return -1;
 195
 196       if (errno == EAGAIN || errno == EWOULDBLOCK) {
 197           if (dsi_peek(dsi)) {
 198               /* can't go back to blocking mode, exit, the next read
 199                  will return with an error and afpd will die.
 200               */
 201               break;
 202           }
 203           continue;
 204       }
 205       LOG(log_error, logtype_dsi, "dsi_stream_read_file: %s", strerror(errno));
 206       break;
 207     }
 208     else if (!len) {
 209         /* afpd is going to exit */
 210         errno = EIO;
 211         return -1; /* I think we're at EOF here... */
 212     }
 213     else
 214         written += len;
 215   }
 216
 217   dsi->write_count += written;
 218   dsi->in_write--;
 219   return written;
 220 }
 221 #endif
 222
 223 /*
 224  * Return all bytes up to count from dsi->buffer if there are any buffered there
 225  */
 226 static size_t from_buf(DSI *dsi, u_int8_t *buf, size_t count)
 227 {
 228     size_t nbe = 0;
 229
 230     LOG(log_maxdebug, logtype_dsi, "from_buf: %u bytes", count);
 231
 232     if (dsi->start) {
 233         nbe = dsi->eof - dsi->start;
 234
 235         if (nbe > 0) {
 236            nbe = min((size_t)nbe, count);
 237            memcpy(buf, dsi->start, nbe);
 238            dsi->start += nbe;
 239
 240            if (dsi->eof == dsi->start)
 241                dsi->start = dsi->eof = dsi->buffer;
 242
 243         }
 244     }
 245
 246     LOG(log_maxdebug, logtype_dsi, "from_buf(dead: %u, unread:%u , space left: %u): returning %u",
 247         dsi->start - dsi->buffer, dsi->eof - dsi->start, dsi->end - dsi->eof, nbe);
 248     return nbe;
 249 }
 250
 251 /*
 252  * Get bytes from buffer dsi->buffer or read from socket
 253  *
 254  * 1. Check if there are bytes in the the dsi->buffer buffer.
 255  * 2. Return bytes from (1) if yes.
 256  *    Note: this may return fewer bytes then requested in count !!
 257  * 3. If the buffer was empty, read from the socket.
 258  */
 259 static ssize_t buf_read(DSI *dsi, u_int8_t *buf, size_t count)
 260 {
 261     ssize_t len;
 262
 263     LOG(log_maxdebug, logtype_dsi, "buf_read(%u bytes)", count);
 264
 265     if (!count)
 266         return 0;
 267
 268     len = from_buf(dsi, buf, count); /* 1. */
 269     if (len)
 270         return len;             /* 2. */
 271
 272     len = readt(dsi->socket, buf, count, 0, 1); /* 3. */
 273
 274     LOG(log_maxdebug, logtype_dsi, "buf_read(%u bytes): got: %d", count, len);
 275
 276     return len;
 277 }
 278
 279 /*
 280  * Essentially a loop around buf_read() to ensure "length" bytes are read
 281  * from dsi->buffer and/or the socket.
 282  */
 283 size_t dsi_stream_read(DSI *dsi, void *data, const size_t length)
 284 {
 285   size_t stored;
 286   ssize_t len;
 287
 288   LOG(log_maxdebug, logtype_dsi, "dsi_stream_read(%u bytes)", length);
 289
 290   stored = 0;
 291   while (stored < length) {
 292       len = buf_read(dsi, (u_int8_t *) data + stored, length - stored);
 293       if (len == -1 && (errno == EINTR || errno == EAGAIN)) {
 294           LOG(log_debug, logtype_dsi, "dsi_stream_read: select read loop");
 295           continue;
 296       } else if (len > 0) {
 297           stored += len;
 298       } else { /* eof or error */
 299           /* don't log EOF error if it's just after connect (OSX 10.3 probe) */
 300           if (len || stored || dsi->read_count) {
 301               if (! (dsi->flags & DSI_DISCONNECTED)) {
 302                   LOG(log_error, logtype_dsi, "dsi_stream_read: len:%d, %s",
 303                       len, (len < 0) ? strerror(errno) : "unexpected EOF");
 304                   AFP_PANIC("FIXME");
 305               }
 306           }
 307           break;
 308       }
 309   }
 310
 311   dsi->read_count += stored;
 312
 313   LOG(log_maxdebug, logtype_dsi, "dsi_stream_read(%u bytes): got: %u", length, stored);
 314   return stored;
 315 }
 316
 317 /*
 318  * Get "length" bytes from buffer and/or socket. In order to avoid frequent small reads
 319  * this tries to read larger chunks (65536 bytes) into a buffer.
 320  */
 321 static size_t dsi_buffered_stream_read(DSI *dsi, u_int8_t *data, const size_t length)
 322 {
 323   size_t len;
 324   size_t buflen;
 325
 326   LOG(log_maxdebug, logtype_dsi, "dsi_buffered_stream_read: %u bytes", length);
 327
 328   len = from_buf(dsi, data, length); /* read from buffer dsi->buffer */
 329   dsi->read_count += len;
 330   if (len == length) {          /* got enough bytes from there ? */
 331       return len;               /* yes */
 332   }
 333
 334   /* fill the buffer with 65536 bytes or until buffer is full */
 335   buflen = min(65536, dsi->end - dsi->eof);
 336   if (buflen > 0) {
 337       ssize_t ret;
 338       ret = read(dsi->socket, dsi->eof, buflen);
 339       if (ret > 0)
 340           dsi->eof += ret;
 341   }
 342
 343   /* now get the remaining data */
 344   len += dsi_stream_read(dsi, data + len, length - len);
 345   return len;
 346 }
 347
 348 /* ---------------------------------------
 349 */
 350 static void block_sig(DSI *dsi)
 351 {
 352   dsi->in_write++;
 353 }
 354
 355 /* ---------------------------------------
 356 */
 357 static void unblock_sig(DSI *dsi)
 358 {
 359   dsi->in_write--;
 360 }
 361
 362 /* ---------------------------------------
 363  * write data. 0 on failure. this assumes that dsi_len will never
 364  * cause an overflow in the data buffer.
 365  */
 366 int dsi_stream_send(DSI *dsi, void *buf, size_t length)
 367 {
 368   char block[DSI_BLOCKSIZ];
 369   struct iovec iov[2];
 370   size_t towrite;
 371   ssize_t len;
 372
 373   LOG(log_maxdebug, logtype_dsi, "dsi_stream_send: %u bytes",
 374       length ? length : sizeof(block));
 375
 376   block[0] = dsi->header.dsi_flags;
 377   block[1] = dsi->header.dsi_command;
 378   memcpy(block + 2, &dsi->header.dsi_requestID,
 379          sizeof(dsi->header.dsi_requestID));
 380   memcpy(block + 4, &dsi->header.dsi_code, sizeof(dsi->header.dsi_code));
 381   memcpy(block + 8, &dsi->header.dsi_len, sizeof(dsi->header.dsi_len));
 382   memcpy(block + 12, &dsi->header.dsi_reserved,
 383          sizeof(dsi->header.dsi_reserved));
 384
 385   if (!length) { /* just write the header */
 386     length = (dsi_stream_write(dsi, block, sizeof(block), 0) == sizeof(block));
 387     return length; /* really 0 on failure, 1 on success */
 388   }
 389
 390   /* block signals */
 391   block_sig(dsi);
 392   iov[0].iov_base = block;
 393   iov[0].iov_len = sizeof(block);
 394   iov[1].iov_base = buf;
 395   iov[1].iov_len = length;
 396
 397   towrite = sizeof(block) + length;
 398   dsi->write_count += towrite;
 399   while (towrite > 0) {
 400       if (((len = writev(dsi->socket, iov, 2)) == -1 && errno == EINTR) || (len == 0))
 401           continue;
 402
 403       if ((size_t)len == towrite) /* wrote everything out */
 404           break;
 405       else if (len < 0) { /* error */
 406           if (errno == EAGAIN || errno == EWOULDBLOCK) {
 407               if (!dsi_peek(dsi)) {
 408                   continue;
 409               }
 410           }
 411           LOG(log_error, logtype_dsi, "dsi_stream_send: %s", strerror(errno));
 412           unblock_sig(dsi);
 413           return 0;
 414       }
 415
 416       towrite -= len;
 417       if (towrite > length) { /* skip part of header */
 418           iov[0].iov_base = (char *) iov[0].iov_base + len;
 419           iov[0].iov_len -= len;
 420       } else { /* skip to data */
 421           if (iov[0].iov_len) {
 422               len -= iov[0].iov_len;
 423               iov[0].iov_len = 0;
 424           }
 425           iov[1].iov_base = (char *) iov[1].iov_base + len;
 426           iov[1].iov_len -= len;
 427       }
 428   }
 429
 430   unblock_sig(dsi);
 431   return 1;
 432 }
 433
 434
 435 /* ---------------------------------------
 436  * read data. function on success. 0 on failure. data length gets
 437  * stored in length variable. this should really use size_t's, but
 438  * that would require changes elsewhere. */
 439 int dsi_stream_receive(DSI *dsi, void *buf, const size_t ilength,
 440                        size_t *rlength)
 441 {
 442   char block[DSI_BLOCKSIZ];
 443
 444   LOG(log_maxdebug, logtype_dsi, "dsi_stream_receive: %u bytes", ilength);
 445
 446   /* read in the header */
 447   if (dsi_buffered_stream_read(dsi, (u_int8_t *)block, sizeof(block)) != sizeof(block))
 448     return 0;
 449
 450   dsi->header.dsi_flags = block[0];
 451   dsi->header.dsi_command = block[1];
 452   /* FIXME, not the right place,
 453      but we get a server disconnect without reason in the log
 454   */
 455   if (!block[1]) {
 456       LOG(log_error, logtype_dsi, "dsi_stream_receive: invalid packet, fatal");
 457       return 0;
 458   }
 459
 460   memcpy(&dsi->header.dsi_requestID, block + 2,
 461          sizeof(dsi->header.dsi_requestID));
 462   memcpy(&dsi->header.dsi_code, block + 4, sizeof(dsi->header.dsi_code));
 463   memcpy(&dsi->header.dsi_len, block + 8, sizeof(dsi->header.dsi_len));
 464   memcpy(&dsi->header.dsi_reserved, block + 12,
 465          sizeof(dsi->header.dsi_reserved));
 466   dsi->clientID = ntohs(dsi->header.dsi_requestID);
 467
 468   /* make sure we don't over-write our buffers. */
 469   *rlength = min(ntohl(dsi->header.dsi_len), ilength);
 470   if (dsi_stream_read(dsi, buf, *rlength) != *rlength)
 471     return 0;
 472
 473   return block[1];
 474 }