Introduce a ring buffer implementation for server's network input. As a
result, clients are not disconnected due to sending large messages.
This commit is contained in:
mmdolze
@@ -2,9 +2,6 @@ Known Bugs:
|
||||
--------------
|
||||
|
||||
V0.5dev:
|
||||
- A client gets disconnected if it sends a many commands quickly. This can
|
||||
happen during initial screen setup or if vBars are used on large (40x4)
|
||||
displays.
|
||||
- If vBars are used together with ICON_BLOCK_FILLED on the same screen and
|
||||
that icon is implemented as a custom char, it sometimes is replaced with
|
||||
a non full block from the vBar.
|
||||
|
||||
@@ -20,6 +20,7 @@ v.0.5dev (ongoing development)
|
||||
+ hd44780: Add a 'none' charmap which does not replace any character
|
||||
* hd44780: Change mapping for spanish 'n with tilde' characters
|
||||
* hd44780: Exclude pin for switchable backlight from keypad scanning
|
||||
* server core: New network input buffering
|
||||
|
||||
v0.5.3
|
||||
+ lcdexec: notification when called program finishes
|
||||
|
||||
+4
-19
@@ -129,33 +129,18 @@ int
|
||||
client_add_message(Client *c, char *message)
|
||||
{
|
||||
int err = 0;
|
||||
char *dup;
|
||||
char *str, *cp;
|
||||
char delimiters[] = "\n\r\0";
|
||||
|
||||
debug(RPT_DEBUG, "%s(c=[%d], message=\"%s\")", __FUNCTION__, c->sock, message);
|
||||
|
||||
if (!c)
|
||||
return -1;
|
||||
if (!message)
|
||||
return -1;
|
||||
|
||||
/* Copy the string to avoid overwriting the original...*/
|
||||
dup = strdup(message);
|
||||
if (!dup) {
|
||||
report(RPT_ERR, "%s: Error allocating", __FUNCTION__);
|
||||
return -1;
|
||||
}
|
||||
/* Now split the string into lines and enqueue each one...*/
|
||||
for (str = strtok(dup, delimiters); str; str = strtok(NULL, delimiters)) {
|
||||
cp = strdup(str);
|
||||
debug(RPT_DEBUG, "%s: Queued message: \"%s\"", __FUNCTION__, cp);
|
||||
err += LL_Enqueue(c->messages, (void *) cp);
|
||||
if (strlen(message) > 0) {
|
||||
debug(RPT_DEBUG, "%s(c=[%d], message=\"%s\")", __FUNCTION__,
|
||||
c->sock, message);
|
||||
err = LL_Enqueue(c->messages, (void *) message);
|
||||
}
|
||||
|
||||
free(dup);
|
||||
|
||||
/* Err is the number of errors encountered...*/
|
||||
return err;
|
||||
}
|
||||
|
||||
|
||||
+58
-58
@@ -11,6 +11,7 @@
|
||||
* 2003, Benjamin Tse (blt@ieee.org) - Winsock port
|
||||
* 2004, F5 Networks, Inc. - IP-address input
|
||||
* 2005, Peter Marschall - error checks, ...
|
||||
* 2009, Markus Dolze - input ring buffer
|
||||
*/
|
||||
|
||||
#ifdef HAVE_CONFIG_H
|
||||
@@ -43,6 +44,8 @@
|
||||
#include "screen.h"
|
||||
#include "shared/report.h"
|
||||
#include "screenlist.h"
|
||||
#include "shared/sring.h"
|
||||
#include "shared/defines.h"
|
||||
|
||||
|
||||
/****************************************************************************/
|
||||
@@ -56,6 +59,9 @@ static int listening_fd;
|
||||
static LinkedList* openSocketList = NULL;
|
||||
static LinkedList* freeClientSocketList = NULL;
|
||||
|
||||
/* ring buffer for incoming messages */
|
||||
static sring_buffer *messageRing;
|
||||
|
||||
/** Mapping between socket and associated client */
|
||||
typedef struct _ClientSocketMap
|
||||
{
|
||||
@@ -147,25 +153,15 @@ sock_init(char* bind_addr, int bind_port)
|
||||
LL_AddNode(openSocketList, (void*) entry);
|
||||
}
|
||||
|
||||
if ((messageRing = sring_create(MAXMSG)) == NULL) {
|
||||
report(RPT_ERR, "%s: error allocating receive buffer.",
|
||||
__FUNCTION__);
|
||||
return -1;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
This code gets the send and receive buffer sizes.
|
||||
{
|
||||
int val, len, sock;
|
||||
sock = new;
|
||||
|
||||
len = sizeof(int);
|
||||
getsockopt(sock, SOL_SOCKET, SO_SNDBUF, &val, &len);
|
||||
debug(RPT_DEBUG, "SEND buffer: %i bytes", val);
|
||||
|
||||
len = sizeof(int);
|
||||
getsockopt(sock, SOL_SOCKET, SO_RCVBUF, &val, &len);
|
||||
debug(RPT_DEBUG, "RECV buffer: %i bytes", val);
|
||||
}
|
||||
*/
|
||||
|
||||
|
||||
/** Cleanup socket management structures.
|
||||
* \retval <0 error
|
||||
@@ -200,6 +196,7 @@ sock_shutdown(void)
|
||||
close(listening_fd);
|
||||
LL_Destroy(freeClientSocketList);
|
||||
free(freeClientSocketPool);
|
||||
sring_destroy(messageRing);
|
||||
|
||||
#ifdef WINSOCK2
|
||||
if (WSACleanup() != 0) {
|
||||
@@ -373,14 +370,11 @@ sock_poll_clients(void)
|
||||
}
|
||||
else { /* Data arriving on an already-connected socket. */
|
||||
int err = 0;
|
||||
|
||||
do {
|
||||
debug(RPT_DEBUG, "%s: reading...", __FUNCTION__);
|
||||
err = sock_read_from_client(clientSocket);
|
||||
debug(RPT_DEBUG, "%s: ...done", __FUNCTION__);
|
||||
if (err < 0)
|
||||
sock_destroy_socket();
|
||||
} while (err > 0);
|
||||
debug(RPT_DEBUG, "%s: reading...", __FUNCTION__);
|
||||
err = sock_read_from_client(clientSocket);
|
||||
debug(RPT_DEBUG, "%s: ...done", __FUNCTION__);
|
||||
if (err < 0)
|
||||
sock_destroy_socket();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -395,44 +389,52 @@ sock_poll_clients(void)
|
||||
static int
|
||||
sock_read_from_client(ClientSocketMap *clientSocketMap)
|
||||
{
|
||||
char buffer[MAXMSG + 1];
|
||||
int nbytes, i;
|
||||
char buffer[MAXMSG];
|
||||
int nbytes;
|
||||
|
||||
debug(RPT_DEBUG, "%s()", __FUNCTION__);
|
||||
|
||||
errno = 0;
|
||||
nbytes = sock_recv(clientSocketMap->socket, buffer, MAXMSG);
|
||||
if (nbytes < 0) {
|
||||
if (errno != EAGAIN)
|
||||
report(RPT_DEBUG, "%s: Error on socket %d - %s",
|
||||
__FUNCTION__, clientSocketMap->socket, sock_geterror());
|
||||
return 0;
|
||||
}
|
||||
else if (nbytes == 0) { /* EOF*/
|
||||
return -1;
|
||||
}
|
||||
else if (nbytes > (MAXMSG - (MAXMSG / 8))) { /* Very noisy client...*/
|
||||
sock_send_error(clientSocketMap->socket, "Too much data received... quiet down!\n");
|
||||
return -1;
|
||||
}
|
||||
else { /* Data Read */
|
||||
buffer[nbytes] = '\0';
|
||||
/* Now, replace zeros with linefeeds...*/
|
||||
for (i = 0; i < nbytes; i++)
|
||||
if (buffer[i] == 0)
|
||||
buffer[i] = '\n';
|
||||
/* Enqueue a "client message" here...*/
|
||||
if (clientSocketMap->client) {
|
||||
client_add_message(clientSocketMap->client, buffer);
|
||||
} else {
|
||||
report(RPT_DEBUG, "%s: Can't find client %d",
|
||||
__FUNCTION__, clientSocketMap->socket);
|
||||
}
|
||||
nbytes = sock_recv(clientSocketMap->socket, buffer, MAXMSG);
|
||||
|
||||
report(RPT_DEBUG, "%s: got message from client %d: \"%s\"",
|
||||
__FUNCTION__, clientSocketMap->socket, buffer);
|
||||
return nbytes;
|
||||
while (nbytes > 0) { /* Data available */
|
||||
int fr;
|
||||
char *str;
|
||||
|
||||
debug(RPT_DEBUG, "%s: received %4d bytes", __FUNCTION__, nbytes);
|
||||
|
||||
/* Append to ring buffer */
|
||||
sring_write(messageRing, buffer, nbytes);
|
||||
|
||||
/* Process all available message in ring buffer */
|
||||
do {
|
||||
str = sring_read_string(messageRing);
|
||||
if (clientSocketMap->client) {
|
||||
client_add_message(clientSocketMap->client, str);
|
||||
} else {
|
||||
report(RPT_DEBUG, "%s: Can't find client %d",
|
||||
__FUNCTION__, clientSocketMap->socket);
|
||||
}
|
||||
} while (str != NULL);
|
||||
|
||||
/* Read again, but only as much as space is left */
|
||||
fr = sring_getMaxWrite(messageRing);
|
||||
if (fr == 0)
|
||||
report(RPT_WARNING, "%s: Message buffer full", __FUNCTION__);
|
||||
|
||||
nbytes = sock_recv(clientSocketMap->socket, buffer, min(MAXMSG, fr));
|
||||
}
|
||||
|
||||
if (sring_getMaxRead(messageRing) > 0) {
|
||||
report(RPT_WARNING, "%s: left over bytes in message buffer",
|
||||
__FUNCTION__);
|
||||
sring_clear(messageRing);
|
||||
}
|
||||
|
||||
if (nbytes < 0 && errno == EAGAIN)
|
||||
return 0; /* No data is not an error */
|
||||
|
||||
return -1; /* EOF */
|
||||
}
|
||||
|
||||
|
||||
@@ -521,5 +523,3 @@ int verify_ipv6(const char *addr)
|
||||
}
|
||||
return (result > 0) ? 1 : 0;
|
||||
}
|
||||
|
||||
|
||||
|
||||
+1
-1
@@ -2,7 +2,7 @@
|
||||
|
||||
noinst_LIBRARIES = libLCDstuff.a
|
||||
|
||||
libLCDstuff_a_SOURCES = LL.c LL.h sockets.c sockets.h str.c str.h configfile.c configfile.h debug.h report.c report.h snprintf.c snprintf.h
|
||||
libLCDstuff_a_SOURCES = LL.c LL.h sockets.c sockets.h str.c str.h configfile.c configfile.h debug.h report.c report.h snprintf.c snprintf.h sring.c sring.h
|
||||
|
||||
libLCDstuff_a_LIBADD = @LIBOBJS@
|
||||
|
||||
|
||||
+271
@@ -0,0 +1,271 @@
|
||||
/** \file shared/sring.c
|
||||
* Circular buffer implementation for string processing.
|
||||
*
|
||||
* \todo Implement sring_peek() and sring_skip().
|
||||
*/
|
||||
|
||||
/*-
|
||||
* This file is part of LCDd, the lcdproc server.
|
||||
*
|
||||
* This file is released under the GNU General Public License.
|
||||
* Refer to the COPYING file distributed with this package.
|
||||
*
|
||||
* Copyright (c) 2009, Markus Dolze
|
||||
*/
|
||||
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#ifdef DEBUG
|
||||
# include <stdio.h>
|
||||
# include <ctype.h>
|
||||
#endif
|
||||
|
||||
#include "sring.h"
|
||||
|
||||
/**
|
||||
* Allocate a new ring buffer data structure.
|
||||
* As this ring buffer is implemented using the 'Always Keep One Byte Open'
|
||||
* strategy, the internal data buffer is (iSize+1) large.
|
||||
*
|
||||
* \param iSize Initial size of the ring buffer
|
||||
* \return Pointer to the created ring buffer
|
||||
*/
|
||||
sring_buffer*
|
||||
sring_create(int iSize)
|
||||
{
|
||||
sring_buffer *buf;
|
||||
|
||||
if ((buf = malloc(sizeof(*buf))) == NULL)
|
||||
return NULL;
|
||||
|
||||
if ((buf->data = malloc(iSize + 1)) == NULL)
|
||||
return NULL;
|
||||
|
||||
buf->size = iSize + 1;
|
||||
buf->w = 0;
|
||||
buf->r = 0;
|
||||
|
||||
return buf;
|
||||
}
|
||||
|
||||
/**
|
||||
* Free memory used by ring buffer.
|
||||
* \param buf Ring buffer to work on
|
||||
*/
|
||||
void
|
||||
sring_destroy(sring_buffer *buf)
|
||||
{
|
||||
if (buf == NULL)
|
||||
return;
|
||||
|
||||
free(buf->data);
|
||||
buf->data = NULL;
|
||||
free(buf);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clears the internal ring buffer.
|
||||
* Existing data is overwritten with NUL bytes.
|
||||
*
|
||||
* \param buf Ring buffer to work on
|
||||
*/
|
||||
void
|
||||
sring_clear(sring_buffer *buf)
|
||||
{
|
||||
if (buf == NULL)
|
||||
return;
|
||||
|
||||
buf->w = 0;
|
||||
buf->r = 0;
|
||||
memset(buf->data, '\0', buf->size);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of bytes that can be written.
|
||||
* \param buf Ring buffer to work on
|
||||
* \return Byte count
|
||||
*/
|
||||
int
|
||||
sring_getMaxWrite(sring_buffer *buf)
|
||||
{
|
||||
int nBytes;
|
||||
|
||||
if (buf == NULL)
|
||||
return 0;
|
||||
|
||||
/* Use 'Always Keep One Byte Open' strategy */
|
||||
if (buf->w < buf->r)
|
||||
nBytes = buf->r - buf->w - 1;
|
||||
else
|
||||
nBytes = (buf->size - buf->w) + buf->r - 1;
|
||||
|
||||
return nBytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of bytes that can be read.
|
||||
* \param buf Ring buffer to work on
|
||||
* \return Byte count
|
||||
*/
|
||||
int
|
||||
sring_getMaxRead(sring_buffer *buf)
|
||||
{
|
||||
int nBytes;
|
||||
|
||||
if (buf == NULL)
|
||||
return 0;
|
||||
|
||||
if (buf->r <= buf->w)
|
||||
nBytes = buf->w - buf->r;
|
||||
else
|
||||
nBytes = (buf->size - buf->r) + buf->w;
|
||||
|
||||
return nBytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write src_len bytes from src into ring buffer.
|
||||
* Fails if not all bytes can be written.
|
||||
*
|
||||
* \param buf Ring buffer to work on
|
||||
* \param src Pointer to source buffer
|
||||
* \param src_len Number of bytes to write at most
|
||||
* \return -1 if not all bytes can be written, 0 otherwise
|
||||
*/
|
||||
int
|
||||
sring_write(sring_buffer *buf, char *src, int src_len)
|
||||
{
|
||||
if (buf == NULL || src == NULL || src_len <= 0)
|
||||
return -1;
|
||||
|
||||
/* XXX: Modify it to write as much as possible? */
|
||||
if (src_len > sring_getMaxWrite(buf))
|
||||
return -1;
|
||||
|
||||
if (buf->w + src_len < buf->size) {
|
||||
memcpy(buf->data + buf->w, src, src_len);
|
||||
buf->w += src_len;
|
||||
}
|
||||
else {
|
||||
int firstBlockLen = buf->size - buf->w;
|
||||
int secondBlockLen = src_len - firstBlockLen;
|
||||
|
||||
memcpy(buf->data + buf->w, src, firstBlockLen);
|
||||
memcpy(buf->data, src + firstBlockLen, secondBlockLen);
|
||||
|
||||
buf->w = secondBlockLen;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read dst_len bytes from ring buffer into destination.
|
||||
* Fails if buffer does not contains dst_len bytes to read from. The target
|
||||
* buffer must be allocated by the application before calling this function.
|
||||
*
|
||||
* \param buf Ring buffer to work on
|
||||
* \param dst Pointer to target buffer
|
||||
* \param dst_len Number of bytes to read at most
|
||||
* \return The number of bytes actually read
|
||||
*/
|
||||
int
|
||||
sring_read(sring_buffer *buf, char *dst, int dst_len)
|
||||
{
|
||||
if (buf == NULL || dst == NULL || dst_len <= 0)
|
||||
return -1;
|
||||
|
||||
/* Do not read more than available */
|
||||
if (dst_len > sring_getMaxRead(buf))
|
||||
dst_len = sring_getMaxRead(buf);
|
||||
|
||||
if (buf->r + dst_len < buf->size) {
|
||||
memcpy(dst, buf->data + buf->r, dst_len);
|
||||
buf->r += dst_len;
|
||||
}
|
||||
else {
|
||||
int firstBlockLen = buf->size - buf->r;
|
||||
int secondBlockLen = dst_len - firstBlockLen;
|
||||
|
||||
memcpy(dst, buf->data + buf->r, firstBlockLen);
|
||||
|
||||
if (secondBlockLen > 0)
|
||||
memcpy(dst + firstBlockLen, buf->data, secondBlockLen);
|
||||
|
||||
buf->r = secondBlockLen;
|
||||
}
|
||||
|
||||
return dst_len;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the next string from the ring buffer.
|
||||
* The next string is a sequence of bytes terminated by \\r, \\n or \\0. The
|
||||
* memory for the string is allocated dynamically and must be free'd by the
|
||||
* application. The string is always NUL terminated, but does not include the
|
||||
* end character.
|
||||
*
|
||||
* \param buf Ring buffer to work on
|
||||
* \return Pointer to allocated string, NULL if no string is available
|
||||
*/
|
||||
char *
|
||||
sring_read_string(sring_buffer *buf)
|
||||
{
|
||||
int n;
|
||||
char *border;
|
||||
char *p;
|
||||
char *dst;
|
||||
int dst_len;
|
||||
|
||||
if (buf == NULL)
|
||||
return NULL;
|
||||
|
||||
n = sring_getMaxRead(buf);
|
||||
border = buf->data + buf->size;
|
||||
p = buf->data + buf->r;
|
||||
|
||||
while (--n >= 0) {
|
||||
if (*p == '\r' || *p == '\n' || *p == '\0')
|
||||
break;
|
||||
p++;
|
||||
if (p == border)
|
||||
p = buf->data;
|
||||
};
|
||||
|
||||
if (n == -1)
|
||||
return NULL;
|
||||
|
||||
dst_len = sring_getMaxRead(buf) - n;
|
||||
if ((dst = malloc(dst_len)) == NULL)
|
||||
return NULL;
|
||||
|
||||
sring_read(buf, dst, dst_len);
|
||||
dst[dst_len-1] = '\0';
|
||||
|
||||
return dst;
|
||||
}
|
||||
|
||||
/**
|
||||
* Print content of buffer to stdout.
|
||||
* Only enabled, if DEBUG is defined.
|
||||
*
|
||||
* \param buf Ring buffer to work on
|
||||
*/
|
||||
void
|
||||
sring_dump(sring_buffer *buf)
|
||||
{
|
||||
#ifdef DEBUG
|
||||
int a;
|
||||
|
||||
if (buf == NULL)
|
||||
return;
|
||||
|
||||
for (a = 0; a < buf->size; a++) {
|
||||
if (isprint(buf->data[a]))
|
||||
printf("'%c' ", buf->data[a]);
|
||||
else
|
||||
printf("0x%02X ", buf->data[a]);
|
||||
}
|
||||
printf("\n");
|
||||
#endif
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
#ifndef SRING_H
|
||||
#define SRING_H
|
||||
|
||||
/** Ring buffer data structure */
|
||||
typedef struct sring_buffer_t {
|
||||
char *data; /**< Dynamically allocated data storage */
|
||||
unsigned int size; /**< The buffer's size */
|
||||
unsigned int w; /**< write pointer */
|
||||
unsigned int r; /**< read pointer */
|
||||
} sring_buffer;
|
||||
|
||||
sring_buffer* sring_create(int iSize);
|
||||
void sring_destroy(sring_buffer *buf);
|
||||
void sring_clear(sring_buffer *buf);
|
||||
int sring_getMaxWrite(sring_buffer *buf);
|
||||
int sring_getMaxRead(sring_buffer *buf);
|
||||
int sring_write(sring_buffer *buf, char *src, int src_len);
|
||||
int sring_read(sring_buffer *buf, char *dst, int dst_len);
|
||||
char* sring_read_string(sring_buffer *buf);
|
||||
void sring_dump(sring_buffer *buf);
|
||||
|
||||
#endif
|
||||
Reference in New Issue
Block a user