projects
/
NeonServV5.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
fixed cmd_neonserv_set trigger mysql query
[NeonServV5.git]
/
src
/
IOHandler.c
diff --git
a/src/IOHandler.c
b/src/IOHandler.c
index 52398de575559823fa70466926377ad5a5066bdc..81d8d0982fd70c9d8db824427baed4acf7fba361 100644
(file)
--- a/
src/IOHandler.c
+++ b/
src/IOHandler.c
@@
-26,6
+26,7
@@
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <winsock2.h>
#include <ws2tcpip.h>
#else
+#include <signal.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
@@
-35,12
+36,21
@@
#include <netinet/tcp.h>
#endif
#include <netinet/tcp.h>
#endif
+#ifndef EWOULDBLOCK
+#define EWOULDBLOCK EAGAIN
+#endif
+
#define MAXLOG 1024
iohandler_log_callback *iolog_backend = NULL;
struct IODescriptor *first_descriptor = NULL;
struct IODescriptor *timer_priority = NULL;
#define MAXLOG 1024
iohandler_log_callback *iolog_backend = NULL;
struct IODescriptor *first_descriptor = NULL;
struct IODescriptor *timer_priority = NULL;
+#ifdef HAVE_PTHREAD_H
+static pthread_mutex_t io_thread_sync;
+static pthread_mutex_t io_poll_sync;
+#endif
+
void iohandler_log(enum IOLogType type, char *text, ...) {
va_list arg_list;
char logBuf[MAXLOG+1];
void iohandler_log(enum IOLogType type, char *text, ...) {
va_list arg_list;
char logBuf[MAXLOG+1];
@@
-67,6
+77,9
@@
struct IOEngine *engine = NULL;
static void iohandler_init_engine() {
if(engine) return;
static void iohandler_init_engine() {
if(engine) return;
+ IOTHREAD_MUTEX_INIT(io_thread_sync);
+ IOTHREAD_MUTEX_INIT(io_poll_sync);
+
//try other engines
if(!engine && engine_kevent.init && engine_kevent.init())
engine = &engine_kevent;
//try other engines
if(!engine && engine_kevent.init && engine_kevent.init())
engine = &engine_kevent;
@@
-88,6
+101,7
@@
static void iohandler_init_engine() {
}
static void iohandler_append(struct IODescriptor *descriptor) {
}
static void iohandler_append(struct IODescriptor *descriptor) {
+ IOSYNCHRONIZE(io_thread_sync);
struct timeval *timeout = ((descriptor->timeout.tv_sec || descriptor->timeout.tv_usec) ? &descriptor->timeout : NULL);
if(timeout) {
struct IODescriptor *iofd;
struct timeval *timeout = ((descriptor->timeout.tv_sec || descriptor->timeout.tv_usec) ? &descriptor->timeout : NULL);
if(timeout) {
struct IODescriptor *iofd;
@@
-102,9
+116,9
@@
static void iohandler_append(struct IODescriptor *descriptor) {
if(timeval_is_smaler(timeout, (&iofd->timeout))) {
descriptor->prev = iofd->prev;
descriptor->next = iofd;
if(timeval_is_smaler(timeout, (&iofd->timeout))) {
descriptor->prev = iofd->prev;
descriptor->next = iofd;
- iofd->prev = descriptor;
if(iofd->prev)
iofd->prev->next = descriptor;
if(iofd->prev)
iofd->prev->next = descriptor;
+ iofd->prev = descriptor;
if(set_priority)
timer_priority = descriptor;
break;
if(set_priority)
timer_priority = descriptor;
break;
@@
-134,10
+148,12
@@
static void iohandler_append(struct IODescriptor *descriptor) {
first_descriptor->prev = descriptor;
first_descriptor = descriptor;
}
first_descriptor->prev = descriptor;
first_descriptor = descriptor;
}
+ IODESYNCHRONIZE(io_thread_sync);
}
static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
//remove IODescriptor from the list
}
static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
//remove IODescriptor from the list
+ IOSYNCHRONIZE(io_thread_sync);
if(descriptor->prev)
descriptor->prev->next = descriptor->next;
else
if(descriptor->prev)
descriptor->prev->next = descriptor->next;
else
@@
-155,6
+171,7
@@
static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove)
free(descriptor->writebuf.buffer);
iohandler_log(IOLOG_DEBUG, "removed IODescriptor (%d) of type `%s`", descriptor->fd, iohandler_iotype_name(descriptor->type));
free(descriptor);
free(descriptor->writebuf.buffer);
iohandler_log(IOLOG_DEBUG, "removed IODescriptor (%d) of type `%s`", descriptor->fd, iohandler_iotype_name(descriptor->type));
free(descriptor);
+ IODESYNCHRONIZE(io_thread_sync);
}
struct IODescriptor *iohandler_add(int sockfd, enum IOType type, struct timeval *timeout, iohandler_callback *callback) {
}
struct IODescriptor *iohandler_add(int sockfd, enum IOType type, struct timeval *timeout, iohandler_callback *callback) {
@@
-327,8
+344,19
@@
struct IODescriptor *iohandler_connect(const char *hostname, unsigned int port,
dstaddrlen = sizeof(*ip4);
} else
return NULL;
dstaddrlen = sizeof(*ip4);
} else
return NULL;
+ //prevent SIGPIPE
+ #ifndef WIN32
+ #if defined(SO_NOSIGPIPE)
+ {
+ int set = 1;
+ setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
+ }
+ #else
+ signal(SIGPIPE, SIG_IGN);
+ #endif
+ #endif
//make sockfd unblocking
//make sockfd unblocking
-#if defined(F_GETFL)
+
#if defined(F_GETFL)
{
int flags;
flags = fcntl(sockfd, F_GETFL);
{
int flags;
flags = fcntl(sockfd, F_GETFL);
@@
-336,11
+364,11
@@
struct IODescriptor *iohandler_connect(const char *hostname, unsigned int port,
flags = fcntl(sockfd, F_GETFD);
fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
}
flags = fcntl(sockfd, F_GETFD);
fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
}
-#else
+
#else
/* I hope you're using the Win32 backend or something else that
* automatically marks the file descriptor non-blocking...
*/
/* I hope you're using the Win32 backend or something else that
* automatically marks the file descriptor non-blocking...
*/
-#endif
+
#endif
descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
if(!descriptor) {
close(sockfd);
descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
if(!descriptor) {
close(sockfd);
@@
-411,8
+439,19
@@
struct IODescriptor *iohandler_listen(const char *hostname, unsigned int port, i
bind(sockfd, (struct sockaddr*)ip4, sizeof(*ip4));
} else
return NULL;
bind(sockfd, (struct sockaddr*)ip4, sizeof(*ip4));
} else
return NULL;
+ //prevent SIGPIPE
+ #ifndef WIN32
+ #if defined(SO_NOSIGPIPE)
+ {
+ int set = 1;
+ setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
+ }
+ #else
+ signal(SIGPIPE, SIG_IGN);
+ #endif
+ #endif
//make sockfd unblocking
//make sockfd unblocking
-#if defined(F_GETFL)
+
#if defined(F_GETFL)
{
int flags;
flags = fcntl(sockfd, F_GETFL);
{
int flags;
flags = fcntl(sockfd, F_GETFL);
@@
-420,11
+459,11
@@
struct IODescriptor *iohandler_listen(const char *hostname, unsigned int port, i
flags = fcntl(sockfd, F_GETFD);
fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
}
flags = fcntl(sockfd, F_GETFD);
fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
}
-#else
+
#else
/* I hope you're using the Win32 backend or something else that
* automatically marks the file descriptor non-blocking...
*/
/* I hope you're using the Win32 backend or something else that
* automatically marks the file descriptor non-blocking...
*/
-#endif
+
#endif
descriptor = iohandler_add(sockfd, IOTYPE_SERVER, NULL, callback);
if(!descriptor) {
close(sockfd);
descriptor = iohandler_add(sockfd, IOTYPE_SERVER, NULL, callback);
if(!descriptor) {
close(sockfd);
@@
-475,8
+514,8
@@
void iohandler_printf(struct IODescriptor *iofd, const char *text, ...) {
iohandler_send(iofd, sendBuf, pos+1);
}
iohandler_send(iofd, sendBuf, pos+1);
}
-
void
iohandler_try_write(struct IODescriptor *iofd) {
- if(!iofd->writebuf.bufpos) return;
+
static int
iohandler_try_write(struct IODescriptor *iofd) {
+ if(!iofd->writebuf.bufpos) return
0
;
iohandler_log(IOLOG_DEBUG, "write writebuf (%d bytes) to socket (fd: %d)", iofd->writebuf.bufpos, iofd->fd);
int res;
if(iofd->ssl_active)
iohandler_log(IOLOG_DEBUG, "write writebuf (%d bytes) to socket (fd: %d)", iofd->writebuf.bufpos, iofd->fd);
int res;
if(iofd->ssl_active)
@@
-484,14
+523,16
@@
void iohandler_try_write(struct IODescriptor *iofd) {
else
res = send(iofd->fd, iofd->writebuf.buffer, iofd->writebuf.bufpos, 0);
if(res < 0) {
else
res = send(iofd->fd, iofd->writebuf.buffer, iofd->writebuf.bufpos, 0);
if(res < 0) {
- if (errno != EAGAIN
) {
+ if (errno != EAGAIN
&& errno != EWOULDBLOCK)
iohandler_log(IOLOG_ERROR, "could not write to socket (fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
iohandler_log(IOLOG_ERROR, "could not write to socket (fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
- }
+ else
+ res = 0;
} else {
iofd->writebuf.bufpos -= res;
if(iofd->state != IO_CLOSED)
engine->update(iofd);
}
} else {
iofd->writebuf.bufpos -= res;
if(iofd->state != IO_CLOSED)
engine->update(iofd);
}
+ return res;
}
void iohandler_close(struct IODescriptor *iofd) {
}
void iohandler_close(struct IODescriptor *iofd) {
@@
-541,6
+582,7
@@
void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
if(!readable && !writeable) {
callback_event.type = IOEVENT_SSLFAILED;
iofd->state = IO_CLOSED;
if(!readable && !writeable) {
callback_event.type = IOEVENT_SSLFAILED;
iofd->state = IO_CLOSED;
+ engine->update(iofd);
} else {
iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_client_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
iohandler_ssl_client_handshake(iofd);
} else {
iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_client_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
iohandler_ssl_client_handshake(iofd);
@@
-597,7
+639,7
@@
void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
bytes = recv(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos, 0);
}
if(bytes <= 0) {
bytes = recv(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos, 0);
}
if(bytes <= 0) {
- if (errno != EAGAIN) {
+ if (errno != EAGAIN
|| errno != EWOULDBLOCK
) {
iofd->state = IO_CLOSED;
engine->update(iofd);
callback_event.type = IOEVENT_CLOSED;
iofd->state = IO_CLOSED;
engine->update(iofd);
callback_event.type = IOEVENT_CLOSED;
@@
-646,7
+688,14
@@
void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
callback_event.type = IOEVENT_READABLE;
}
if(writeable) {
callback_event.type = IOEVENT_READABLE;
}
if(writeable) {
- iohandler_try_write(iofd);
+ int bytes;
+ bytes = iohandler_try_write(iofd);
+ if(bytes < 0) {
+ iofd->state = IO_CLOSED;
+ engine->update(iofd);
+ callback_event.type = IOEVENT_CLOSED;
+ callback_event.data.errid = errno;
+ }
}
break;
}
}
break;
}
@@
-658,10
+707,12
@@
void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
void iohandler_poll() {
if(engine) {
void iohandler_poll() {
if(engine) {
+ IOSYNCHRONIZE(io_poll_sync); //quite senceless multithread support... better support will follow
struct timeval timeout;
timeout.tv_sec = IO_MAX_TIMEOUT;
timeout.tv_usec = 0;
engine->loop(&timeout);
struct timeval timeout;
timeout.tv_sec = IO_MAX_TIMEOUT;
timeout.tv_usec = 0;
engine->loop(&timeout);
+ IODESYNCHRONIZE(io_poll_sync);
}
}
}
}