/*
- * $Id: dsi_stream.c,v 1.11 2003-03-12 15:07:06 didg Exp $
+ * $Id: dsi_stream.c,v 1.12 2005-04-28 20:50:02 bfernhomberg Exp $
*
* Copyright (c) 1998 Adrian Sun (asun@zoology.washington.edu)
* All rights reserved. See COPYRIGHT.
#include <stdio.h>
#include <stdlib.h>
+
#ifdef HAVE_UNISTD_H
#include <unistd.h>
-#endif /* HAVE_UNISTD_H */
+#endif
+
#include <string.h>
#include <errno.h>
#include <sys/types.h>
+#include <sys/socket.h>
+
#ifdef USE_WRITEV
#include <sys/uio.h>
-#endif /* USE_WRITEV */
+#endif
+
#include <atalk/logger.h>
#include <atalk/dsi.h>
#define min(a,b) ((a) < (b) ? (a) : (b))
+#ifndef MSG_MORE
+#define MSG_MORE 0x8000
+#endif
+
+#ifndef MSG_DONTWAIT
+#define MSG_DONTWAIT 0x40
+#endif
-/* write raw data. return actual bytes read. checks against EINTR
+/* -------------------------
+ * we don't use a circular buffer.
+*/
+void dsi_buffer(DSI *dsi)
+{
+ fd_set readfds, writefds;
+ int len;
+ int maxfd;
+
+ FD_ZERO(&readfds);
+ FD_ZERO(&writefds);
+ FD_SET( dsi->socket, &readfds);
+ FD_SET( dsi->socket, &writefds);
+ maxfd = dsi->socket +1;
+ while (1) {
+ FD_SET( dsi->socket, &readfds);
+ FD_SET( dsi->socket, &writefds);
+ if (select( maxfd, &readfds, &writefds, NULL, NULL) <= 0)
+ return;
+
+ if ( !FD_ISSET(dsi->socket, &readfds)) {
+ /* nothing waiting in the read queue */
+ return;
+ }
+ if (!dsi->buffer) {
+ /* XXX config options */
+ dsi->maxsize = 6 * dsi->server_quantum;
+ if (!dsi->maxsize)
+ dsi->maxsize = 6 * DSI_SERVQUANT_DEF;
+ dsi->buffer = malloc(dsi->maxsize);
+ if (!dsi->buffer) {
+ /* fall back to blocking IO */
+ dsi_block(dsi, 0);
+ return;
+ }
+ dsi->start = dsi->buffer;
+ dsi->eof = dsi->buffer;
+ dsi->end = dsi->buffer + dsi->maxsize;
+ }
+ len = dsi->end - dsi->eof;
+
+ if (len <= 0) {
+ /* ouch, our buffer is full !
+ * fall back to blocking IO
+ * could block and disconnect but it's better than a cpu hog
+ */
+ dsi_block(dsi, 0);
+ return;
+ }
+
+ len = read(dsi->socket, dsi->eof, len);
+ if (len <= 0)
+ return;
+ dsi->eof += len;
+ if ( FD_ISSET(dsi->socket, &writefds)) {
+ return;
+ }
+ }
+}
+
+/* ------------------------------
+ * write raw data. return actual bytes read. checks against EINTR
* aren't necessary if all of the signals have SA_RESTART
* specified. */
-size_t dsi_stream_write(DSI *dsi, void *data, const size_t length)
+size_t dsi_stream_write(DSI *dsi, void *data, const size_t length, int mode _U_)
{
size_t written;
ssize_t len;
+#if 0
+ /* FIXME sometime it's slower */
+ unsigned int flags = (mode)?MSG_MORE:0;
+#endif
+ unsigned int flags = 0;
+#if 0
+ /* XXX there's no MSG_DONTWAIT in recv ?? so we have to play with ioctl
+ */
+ if (dsi->noblocking) {
+ flags |= MSG_DONTWAIT;
+ }
+#endif
+
written = 0;
while (written < length) {
- if ((-1 == (len = write(dsi->socket, (u_int8_t *) data + written,
- length - written)) && errno == EINTR) ||
+ if ((-1 == (len = send(dsi->socket, (u_int8_t *) data + written,
+ length - written, flags)) && errno == EINTR) ||
!len)
continue;
if (len < 0) {
- LOG(log_error, logtype_default, "dsi_stream_write: %s", strerror(errno));
- break;
+ if (dsi->noblocking && errno == EAGAIN) {
+ /* non blocking mode but will block
+ * read data in input queue.
+ *
+ */
+ dsi_buffer(dsi);
+ }
+ else {
+ LOG(log_error, logtype_default, "dsi_stream_write: %s", strerror(errno));
+ break;
+ }
+ }
+ else {
+ written += len;
}
-
- written += len;
}
dsi->write_count += written;
return written;
}
-/* read raw data. return actual bytes read. this will wait until
- * it gets length bytes */
+/* ---------------------------------
+*/
+static ssize_t buf_read(DSI *dsi, u_int8_t *buf, size_t count)
+{
+ ssize_t nbe = 0;
+ ssize_t ret;
+
+ if (!count)
+ return 0;
+
+ if (dsi->start) {
+ nbe = dsi->eof - dsi->start;
+
+ if (nbe > 0) {
+ nbe = min((size_t)nbe, count);
+ memcpy(buf, dsi->start, nbe);
+ dsi->start += nbe;
+
+ if (dsi->eof == dsi->start)
+ dsi->start = dsi->eof = dsi->buffer;
+
+ if (nbe == count)
+ return nbe;
+ count -= nbe;
+ buf += nbe;
+ }
+ else
+ nbe = 0;
+ }
+
+ ret = read(dsi->socket, buf, count);
+ if (ret <= 0)
+ return ret;
+
+ return ret +nbe;
+}
+
+/* ---------------------------------------
+ * read raw data. return actual bytes read. this will wait until
+ * it gets length bytes
+ */
size_t dsi_stream_read(DSI *dsi, void *data, const size_t length)
{
size_t stored;
stored = 0;
while (stored < length) {
- len = read(dsi->socket, (u_int8_t *) data + stored, length - stored);
+ len = buf_read(dsi, (u_int8_t *) data + stored, length - stored);
if (len == -1 && errno == EINTR)
continue;
else if (len > 0)
stored += len;
else { /* eof or error */
- LOG(log_error, logtype_default, "dsi_stream_read(%d): %s", len, (len < 0)?strerror(errno):"unexpected EOF");
+ /* don't log EOF error if it's just after connect (OSX 10.3 probe) */
+ if (len || stored || dsi->read_count) {
+ LOG(log_error, logtype_default, "dsi_stream_read(%d): %s", len, (len < 0)?strerror(errno):"unexpected EOF");
+ }
break;
}
}
return stored;
}
+/* ---------------------------------------
+*/
void dsi_sleep(DSI *dsi, const int state)
{
dsi->asleep = state;
}
-/* write data. 0 on failure. this assumes that dsi_len will never
- * cause an overflow in the data buffer. */
+/* ---------------------------------------
+*/
+static void block_sig(DSI *dsi)
+{
+ if (!dsi->sigblocked) sigprocmask(SIG_BLOCK, &dsi->sigblockset, &dsi->oldset);
+}
+
+/* ---------------------------------------
+*/
+static void unblock_sig(DSI *dsi)
+{
+ if (!dsi->sigblocked) sigprocmask(SIG_SETMASK, &dsi->oldset, NULL);
+}
+
+/* ---------------------------------------
+ * write data. 0 on failure. this assumes that dsi_len will never
+ * cause an overflow in the data buffer.
+ */
int dsi_stream_send(DSI *dsi, void *buf, size_t length)
{
char block[DSI_BLOCKSIZ];
- sigset_t oldset;
#ifdef USE_WRITEV
struct iovec iov[2];
size_t towrite;
memcpy(block + 12, &dsi->header.dsi_reserved,
sizeof(dsi->header.dsi_reserved));
- /* block signals */
- sigprocmask(SIG_BLOCK, &dsi->sigblockset, &oldset);
-
if (!length) { /* just write the header */
- length = (dsi_stream_write(dsi, block, sizeof(block)) == sizeof(block));
- sigprocmask(SIG_SETMASK, &oldset, NULL);
+ length = (dsi_stream_write(dsi, block, sizeof(block), 0) == sizeof(block));
return length; /* really 0 on failure, 1 on success */
}
+ /* block signals */
+ block_sig(dsi);
#ifdef USE_WRITEV
iov[0].iov_base = block;
iov[0].iov_len = sizeof(block);
break;
else if (len < 0) { /* error */
LOG(log_error, logtype_default, "dsi_stream_send: %s", strerror(errno));
- sigprocmask(SIG_SETMASK, &oldset, NULL);
+ unblock_sig(dsi);
return 0;
}
#else /* USE_WRITEV */
/* write the header then data */
- if ((dsi_stream_write(dsi, block, sizeof(block)) != sizeof(block)) ||
- (dsi_stream_write(dsi, buf, length) != length)) {
- sigprocmask(SIG_SETMASK, &oldset, NULL);
- return 0;
+ if ((dsi_stream_write(dsi, block, sizeof(block), 1) != sizeof(block)) ||
+ (dsi_stream_write(dsi, buf, length, 0) != length)) {
+ unblock_sig(dsi);
+ return 0;
}
#endif /* USE_WRITEV */
- sigprocmask(SIG_SETMASK, &oldset, NULL);
+ unblock_sig(dsi);
return 1;
}
-/* read data. function on success. 0 on failure. data length gets
+/* ---------------------------------------
+ * read data. function on success. 0 on failure. data length gets
* stored in length variable. this should really use size_t's, but
* that would require changes elsewhere. */
int dsi_stream_receive(DSI *dsi, void *buf, const size_t ilength,
projects / netatalk.git / blobdiff