src/syncio.c

   1 /* Synchronous socket and file I/O operations useful across the core.
   2  *
   3  * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions are met:
   8  *
   9  *   * Redistributions of source code must retain the above copyright notice,
  10  *     this list of conditions and the following disclaimer.
  11  *   * Redistributions in binary form must reproduce the above copyright
  12  *     notice, this list of conditions and the following disclaimer in the
  13  *     documentation and/or other materials provided with the distribution.
  14  *   * Neither the name of Redis nor the names of its contributors may be used
  15  *     to endorse or promote products derived from this software without
  16  *     specific prior written permission.
  17  *
  18  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  19  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  20  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  21  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  22  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  23  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  24  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  25  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  26  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  27  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  */
  30
  31 #include "redis.h"
  32
  33 /* ----------------- Blocking sockets I/O with timeouts --------------------- */
  34
  35 /* Redis performs most of the I/O in a nonblocking way, with the exception
  36  * of the SYNC command where the slave does it in a blocking way, and
  37  * the MIGRATE command that must be blocking in order to be atomic from the
  38  * point of view of the two instances (one migrating the key and one receiving
  39  * the key). This is why need the following blocking I/O functions.
  40  *
  41  * All the functions take the timeout in milliseconds. */
  42
  43 #define REDIS_SYNCIO_RESOLUTION 10 /* Resolution in milliseconds */
  44
  45 /* Write the specified payload to 'fd'. If writing the whole payload will be done
  46  * within 'timeout' milliseconds the operation succeeds and 'size' is returned.
  47  * Otherwise the operation fails, -1 is returned, and an unspecified partial write
  48  * could be performed against the file descriptor. */
  49 ssize_t syncWrite(int fd, char *ptr, ssize_t size, long long timeout) {
  50     ssize_t nwritten, ret = size;
  51     long long start = mstime();
  52     long long remaining = timeout;
  53
  54     while(1) {
  55         long long wait = (remaining > REDIS_SYNCIO_RESOLUTION) ?
  56                           remaining : REDIS_SYNCIO_RESOLUTION;
  57         long long elapsed;
  58
  59         if (aeWait(fd,AE_WRITABLE,wait) & AE_WRITABLE) {
  60             nwritten = write(fd,ptr,size);
  61             if (nwritten == -1) return -1;
  62             ptr += nwritten;
  63             size -= nwritten;
  64             if (size == 0) return ret;
  65         }
  66         elapsed = mstime() - start;
  67         if (elapsed >= timeout) {
  68             errno = ETIMEDOUT;
  69             return -1;
  70         }
  71         remaining = timeout - elapsed;
  72     }
  73 }
  74
  75 /* Read the specified amount of bytes from 'fd'. If all the bytes are read within
  76  * 'timeout' milliseconds the operation succeed and 'size' is returned.
  77  * Otherwise the operation fails, -1 is returned, and an unspecified amount of
  78  * data could be read from the file descriptor. */
  79 ssize_t syncRead(int fd, char *ptr, ssize_t size, long long timeout) {
  80     ssize_t nread, totread = 0;
  81     long long start = mstime();
  82     long long remaining = timeout;
  83
  84     while(1) {
  85         long long wait = (remaining > REDIS_SYNCIO_RESOLUTION) ?
  86                           remaining : REDIS_SYNCIO_RESOLUTION;
  87         long long elapsed;
  88
  89         if (aeWait(fd,AE_READABLE,wait) & AE_READABLE) {
  90             nread = read(fd,ptr,size);
  91             if (nread <= 0) return -1;
  92             ptr += nread;
  93             size -= nread;
  94             totread += nread;
  95             if (size == 0) return totread;
  96         }
  97         elapsed = mstime() - start;
  98         if (elapsed >= timeout) {
  99             errno = ETIMEDOUT;
 100             return -1;
 101         }
 102         remaining = timeout - elapsed;
 103     }
 104 }
 105
 106 /* Read a line making sure that every char will not require more than 'timeout'
 107  * milliseconds to be read.
 108  *
 109  * On success the number of bytes read is returned, otherwise -1.
 110  * On success the string is always correctly terminated with a 0 byte. */
 111 ssize_t syncReadLine(int fd, char *ptr, ssize_t size, long long timeout) {
 112     ssize_t nread = 0;
 113
 114     size--;
 115     while(size) {
 116         char c;
 117
 118         if (syncRead(fd,&c,1,timeout) == -1) return -1;
 119         if (c == '\n') {
 120             *ptr = '\0';
 121             if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
 122             return nread;
 123         } else {
 124             *ptr++ = c;
 125             *ptr = '\0';
 126             nread++;
 127         }
 128     }
 129     return nread;
 130 }