1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
/* Synchronous socket and file I/O operations useful across the core.
*
* Copyright (c) 2009-Present, Redis Ltd.
* All rights reserved.
*
* Licensed under your choice of (a) the Redis Source Available License 2.0
* (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
* GNU Affero General Public License v3 (AGPLv3).
*/
#include "server.h"
/* ----------------- Blocking sockets I/O with timeouts --------------------- */
/* Redis performs most of the I/O in a nonblocking way, with the exception
* of the SYNC command where the slave does it in a blocking way, and
* the MIGRATE command that must be blocking in order to be atomic from the
* point of view of the two instances (one migrating the key and one receiving
* the key). This is why need the following blocking I/O functions.
*
* All the functions take the timeout in milliseconds. */
#define SYNCIO__RESOLUTION 10 /* Resolution in milliseconds */
/* Write the specified payload to 'fd'. If writing the whole payload will be
* done within 'timeout' milliseconds the operation succeeds and 'size' is
* returned. Otherwise the operation fails, -1 is returned, and an unspecified
* partial write could be performed against the file descriptor. */
ssize_t syncWrite(int fd, char *ptr, ssize_t size, long long timeout) {
ssize_t nwritten, ret = size;
long long start = mstime();
long long remaining = timeout;
while(1) {
long long wait = (remaining > SYNCIO__RESOLUTION) ?
remaining : SYNCIO__RESOLUTION;
long long elapsed;
/* Optimistically try to write before checking if the file descriptor
* is actually writable. At worst we get EAGAIN. */
nwritten = write(fd,ptr,size);
if (nwritten == -1) {
if (errno != EAGAIN) return -1;
} else {
ptr += nwritten;
size -= nwritten;
}
if (size == 0) return ret;
/* Wait */
aeWait(fd,AE_WRITABLE,wait);
elapsed = mstime() - start;
if (elapsed >= timeout) {
errno = ETIMEDOUT;
return -1;
}
remaining = timeout - elapsed;
}
}
/* Read the specified amount of bytes from 'fd'. If all the bytes are read
* within 'timeout' milliseconds the operation succeed and 'size' is returned.
* Otherwise the operation fails, -1 is returned, and an unspecified amount of
* data could be read from the file descriptor. */
ssize_t syncRead(int fd, char *ptr, ssize_t size, long long timeout) {
ssize_t nread, totread = 0;
long long start = mstime();
long long remaining = timeout;
if (size == 0) return 0;
while(1) {
long long wait = (remaining > SYNCIO__RESOLUTION) ?
remaining : SYNCIO__RESOLUTION;
long long elapsed;
/* Optimistically try to read before checking if the file descriptor
* is actually readable. At worst we get EAGAIN. */
nread = read(fd,ptr,size);
if (nread == 0) return -1; /* short read. */
if (nread == -1) {
if (errno != EAGAIN) return -1;
} else {
ptr += nread;
size -= nread;
totread += nread;
}
if (size == 0) return totread;
/* Wait */
aeWait(fd,AE_READABLE,wait);
elapsed = mstime() - start;
if (elapsed >= timeout) {
errno = ETIMEDOUT;
return -1;
}
remaining = timeout - elapsed;
}
}
/* Read a line making sure that every char will not require more than 'timeout'
* milliseconds to be read.
*
* On success the number of bytes read is returned, otherwise -1.
* On success the string is always correctly terminated with a 0 byte. */
ssize_t syncReadLine(int fd, char *ptr, ssize_t size, long long timeout) {
ssize_t nread = 0;
size--;
while(size) {
char c;
if (syncRead(fd,&c,1,timeout) == -1) return -1;
if (c == '\n') {
*ptr = '\0';
if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
return nread;
} else {
*ptr++ = c;
*ptr = '\0';
nread++;
}
size--;
}
return nread;
}
|
