From 0f4e34a006a79fb97d10a4f60ab6f91bd788feae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Thu, 23 May 2013 16:59:59 +0800 Subject: [PATCH] client : put unsend bytes into temp buffer --- service-src/service_client.c | 105 ++++++++++++++++++++++++++++++++--- 1 file changed, 98 insertions(+), 7 deletions(-) diff --git a/service-src/service_client.c b/service-src/service_client.c index 08f61215..a20346c9 100644 --- a/service-src/service_client.c +++ b/service-src/service_client.c @@ -1,6 +1,7 @@ #include "skynet.h" #include +#include #include #include #include @@ -8,10 +9,88 @@ #include #include +struct send_buffer { + struct send_buffer * next; + int size; + char buf[2]; +}; + +struct client { + int fd; + struct send_buffer * head; + struct send_buffer * tail; +}; + +static struct send_buffer * +sb_alloc(size_t sz, const char * msg, int alreadysend) { + struct send_buffer * sb = malloc(sizeof(struct send_buffer) + sz - alreadysend); + if (alreadysend <= 0) { + sb->buf[0] = sz >> 8 & 0xff; + sb->buf[1] = sz & 0xff; + memcpy(sb->buf+2, msg, sz); + sb->size = sz + 2; + } else if (alreadysend == 1) { + sb->buf[0] = sz & 0xff; + memcpy(sb->buf+1, msg, sz); + sb->size = sz + 1; + } else { + sb->size = sz + 2 - alreadysend; + memcpy(sb->buf, msg, sb->size); + } + sb->next = NULL; + return sb; +} + +static void +sb_free_all(struct send_buffer * sb) { + while (sb) { + struct send_buffer * tmp = sb; + sb = sb->next; + free(tmp); + } +} + +static void +send_all(struct client *c) { + while (c->head) { + struct send_buffer * tmp = c->head; + for (;;) { + int sz = write(c->fd, tmp->buf, tmp->size); + if (sz < 0) { + switch(errno) { + case EAGAIN: + case EINTR: + continue; + } + return; + } + if (sz != tmp->size) { + assert(sz < tmp->size); + memmove(tmp->buf, tmp->buf + sz, tmp->size - sz); + tmp->size -= sz; + return; + } + break; + } + c->head = tmp->next; + free(tmp); + } + c->tail = NULL; +} + static int _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { assert(sz <= 65535); - int fd = (int)(intptr_t)ud; + struct client * c = ud; + send_all(c); + if (c->head) { + struct send_buffer * tmp = sb_alloc(sz, msg, 0); + c->tail->next = tmp; + c->tail = tmp; + return 0; + } + + int fd = c->fd; struct iovec buffer[2]; // send big-endian header @@ -30,19 +109,31 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t continue; } } - if (err < 0) { - skynet_error(context, "Client socket error : Drop message from %x session = %d", source, session); - return 0; + if ( err != sz + 2) { + c->head = c->tail = sb_alloc(sz, msg, err); } - assert(err == sz +2); return 0; } } int -client_init(void * dummy, struct skynet_context *ctx, const char * args) { +client_init(struct client *c, struct skynet_context *ctx, const char * args) { int fd = strtol(args, NULL, 10); - skynet_callback(ctx, (void*)(intptr_t)fd, _cb); + c->fd = fd; + skynet_callback(ctx, c, _cb); return 0; } + +struct client * +client_create(void) { + struct client *c = malloc(sizeof(*c)); + memset(c,0,sizeof(*c)); + return c; +} + +void +client_release(struct client *c) { + sb_free_all(c->head); + free(c); +}