static int read_buffer_len;
static int read_buffer_size;
static int no_flush;
+static int no_flush_read;
/* read from a socket with IO timeout. return the number of
bytes read. If no bytes can be read then exit, never return
{
int n, ret=0;
+ no_flush_read++;
io_flush();
+ no_flush_read--;
while (ret == 0) {
fd_set fds;
n = read(fd, buf, len);
if (n > 0) {
- stats.total_read += n;
buf += n;
len -= n;
ret += n;
continue;
}
+ no_flush_read++;
io_flush();
+ no_flush_read--;
ret = read_unbuffered(fd,buffer + total,N-total);
total += ret;
}
+
+ stats.total_read += total;
}
int32 read_int(int f)
{
char b[4];
+ int32 ret;
+
readfd(f,b,4);
- return IVAL(b,0);
+ ret = IVAL(b,0);
+ if (ret == (int32)0xffffffff) return -1;
+ return ret;
}
int64 read_longint(int f)
char b[8];
ret = read_int(f);
- if ((int32)ret != (int32)0xffffffff) return ret;
+ if ((int32)ret != (int32)0xffffffff) {
+ return ret;
+ }
#ifdef NO_INT64
rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
fd_set w_fds, r_fds;
int fd_count, count;
struct timeval tv;
- int reading;
+ int reading=0;
int blocked=0;
no_flush++;
FD_SET(fd,&w_fds);
fd_count = fd+1;
- reading = (buffer_f_in != -1 &&
- read_buffer_len < MAX_READ_BUFFER);
+ if (!no_flush_read) {
+ reading = (buffer_f_in != -1);
+ }
if (reading) {
FD_SET(buffer_f_in,&r_fds);
blocked = 0;
total += ret;
- stats.total_written += ret;
if (io_timeout)
last_io = time(NULL);
static void writefd(int fd,char *buf,int len)
{
+ stats.total_written += len;
+
if (!io_buffer) {
writefd_unbuffered(fd, buf, len);
return;
return;
}
- write_int(f, -1);
+ write_int(f, (int32)0xFFFFFFFF);
SIVAL(b,0,(x&0xFFFFFFFF));
SIVAL(b,4,((x>>32)&0xFFFFFFFF));
}
/* write a string to the connection */
-void write_sbuf(int f,char *buf)
+static void write_sbuf(int f,char *buf)
{
write_buf(f, buf, strlen(buf));
}
SIVAL(io_buffer-4, 0, ((MPLEX_BASE + f)<<24) + len);
memcpy(io_buffer, buf, len);
+ stats.total_written += (len+4);
+
writefd_unbuffered(multiplex_out_fd, io_buffer-4, len+4);
return 1;
}