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