-/* $Id: mmftpd.c,v 1.66 2004/09/19 11:02:45 mmondor Exp $ */
+/* $Id: mmftpd.c,v 1.66.2.1 2005/11/16 01:00:58 mmondor Exp $ */
/*
* Copyright (C) 2001-2004, Matthew Mondor
#include <ctype.h>
#include <syslog.h>
-#include <pth.h>
+#include <pthread.h>
#include <mmtypes.h>
#include <mmreadcfg.h>
#include <mmstring.h>
#include <mmfifo.h>
#include <mmstat.h>
-#include <mm_pth_pool.h>
+
+#include <mm_pthread_msg.h>
+#include <mm_pthread_pool.h>
+#include <mm_pthread_poll.h>
+#include <mm_pthread_sleep.h>
#include "mmftpd.h"
MMCOPYRIGHT("@(#) Copyright (c) 2001-2004\n\
\tMatthew Mondor. All rights reserved.\n");
-MMRCSID("$Id: mmftpd.c,v 1.66 2004/09/19 11:02:45 mmondor Exp $");
+MMRCSID("$Id: mmftpd.c,v 1.66.2.1 2005/11/16 01:00:58 mmondor Exp $");
static CONFIG CONF;
/* Used to start transfer threads */
-static pth_attr_t tthreadattr;
+static pthread_attr_t tthreadattr;
/* List of logged in users and optionally current home directory size */
-static pth_mutex_t lusers_lock;
+static pthread_mutex_t lusers_lock;
static pool_t lusers_pool;
static hashtable_t lusers_table;
/* This is used so that clientenv structures be allocated/freed fast */
-static pth_mutex_t clenvs_lock;
+static pthread_mutex_t clenvs_lock;
static pool_t clenvs_pool;
static pool_t fifos_pool;
/* Pool used to optimize creating/destroying mmfd mutexes */
-static pth_mutex_t mutexes_lock;
+static pthread_mutex_t mutexes_lock;
static pool_t mutexes_pool;
/* Used for the longer-term directory size cache */
-static pth_mutex_t quota_lock;
+static pthread_mutex_t quota_lock;
static pool_t quota_pool;
static hashtable_t quota_table;
static fdfuncs gfdf = {
malloc,
free,
- pth_poll,
- pth_read,
- pth_write,
- pth_sleep,
- pth_usleep,
- _pth_mutex_create,
- _pth_mutex_destroy,
- _pth_mutex_lock,
- _pth_mutex_unlock,
- _pth_thread_yield,
- _pth_eintr
+ poll,
+ read,
+ write,
+ pthread_sleep,
+ pthread_usleep,
+ thread_mutex_create,
+ thread_mutex_destroy,
+ thread_mutex_lock,
+ thread_mutex_unlock,
+ NULL,
+ thread_eintr
};
/* async_getuserline() uses this for each asynchroneous process */
/* We slow down password guessing programs */
if (clenv->attempts > 0)
- pth_sleep(clenv->attempts * 2);
+ pthread_sleep(clenv->attempts * 2);
clenv->attempts++;
if (clenv->anonymous) {
}
directory_message(clenv, 230);
- pth_mutex_acquire(&lusers_lock, FALSE, NULL);
+ pthread_mutex_lock(&lusers_lock);
/* Verify if user in list */
len = mm_strlen(clenv->user) + 1;
bool lok = TRUE;
/* Yes, make sure that we observe limits, if any */
- pth_mutex_acquire(&lun->lock, FALSE, NULL);
+ pthread_mutex_lock(&lun->lock);
if (clenv->maxlogins != 0) {
if (lun->logins < clenv->maxlogins)
lun->logins++;
else lok = FALSE;
} else
lun->logins++;
- pth_mutex_release(&lun->lock);
+ pthread_mutex_unlock(&lun->lock);
if (lok) {
clenv->unode = lun;
if (clenv->anonymous)
nextstate = STATE_ERROR;
}
if (lun != NULL) {
- pth_mutex_init(&lun->lock);
+ pthread_mutex_init(&lun->lock, NULL);
lun->logins = 1;
mm_memcpy(lun->user, clenv->user, len);
clenv->unode = lun;
}
}
- pth_mutex_release(&lusers_lock);
+ pthread_mutex_unlock(&lusers_lock);
}
return (nextstate);
{
long logins;
- pth_mutex_acquire(&clenv->unode->lock, FALSE, NULL);
+ pthread_mutex_lock(&clenv->unode->lock);
logins = clenv->unode->logins;
- pth_mutex_release(&clenv->unode->lock);
+ pthread_mutex_unlock(&clenv->unode->lock);
fdbprintf(fdb, " Logged in as %s (%ld concurrent logins)\r\n",
clenv->user, logins);
}
fdbprintf(fdb, " %lld bytes", (int64_t)clenv->maxhomesize);
fdbwrite(fdb, "\r\n Current home directory size: ", 36);
- pth_mutex_acquire(&clenv->unode->lock, FALSE, NULL);
+ pthread_mutex_lock(&clenv->unode->lock);
size = clenv->unode->homesize;
- pth_mutex_release(&clenv->unode->lock);
+ pthread_mutex_unlock(&clenv->unode->lock);
fdbprintf(fdb, "%lld bytes", (int64_t)size);
} else
fdbwrite(fdb, "disabled", 8);
async_init(afuncs, CONF.ASYNC_PROCESSES, uid, gids, ngids);
/* Threading system initialization */
- pth_init();
+ /* pthread_init(); */
/* Part of mmserver(3)'s async system which has to be initialized after
* the threading system
*/
- async_init_pth();
+ async_init_pthread();
- pth_mutex_init(&lusers_lock);
- pth_mutex_init(&clenvs_lock);
- pth_mutex_init(&mutexes_lock);
- pth_mutex_init("a_lock);
+ pthread_mutex_init(&lusers_lock, NULL);
+ pthread_mutex_init(&clenvs_lock, NULL);
+ pthread_mutex_init(&mutexes_lock, NULL);
+ pthread_mutex_init("a_lock, NULL);
srandom(getpid() + time(NULL));
/* We use those for our transfer ASYNC thread, joinable because when we
* eventually request it to quit via a message we want to make sure it
- * ended properly waiting for it with pth_join(). I have decided to use
+ * ended properly waiting for it with pthread_join(). I have decided to use
* a second thread dedicated to file transfers, this way I can respect
* the RFC on listening to ABOR and STAT easily on the control connection
* while transfers are ongoing, with ease and modularity.
*/
- tthreadattr = pth_attr_new();
- pth_attr_set(tthreadattr, PTH_ATTR_JOINABLE, TRUE);
+ pthread_attr_init(&tthreadattr);
+ pthread_attr_setdetachstate(&tthreadattr, 0);
/* Initialize our pools */
/* Client environment nodes */
POOL_VALID(&mutexes_pool) && POOL_VALID("a_pool) &&
HASHTABLE_VALID("a_table) &&
(!*CONF.DISPLAY_FILE || POOL_VALID(&fifos_pool))) {
+ thread_init();
fdbcinit(&gfdf, &fdbc, CONF.GBANDWIDTH_IN * 1024,
CONF.GBANDWIDTH_OUT * 1024);
/* Finally start our TCP main daemon */
bool ret = TRUE;
if (clenv->maxhomesize && modify != 0) {
- pth_mutex_acquire(&clenv->unode->lock, FALSE, NULL);
+ pthread_mutex_lock(&clenv->unode->lock);
cur = clenv->unode->homesize;
cur += modify;
if (cur < 0)
clenv->unode->homesize = cur;
else
ret = FALSE;
- pth_mutex_release(&clenv->unode->lock);
+ pthread_mutex_unlock(&clenv->unode->lock);
}
return (ret);
clientenv *clenv;
struct fifonode *fnode = NULL;
- pth_mutex_acquire(&clenvs_lock, FALSE, NULL);
+ pthread_mutex_lock(&clenvs_lock);
clenv = (clientenv *)pool_alloc(&clenvs_pool, TRUE);
if (*CONF.DISPLAY_FILE)
fnode = (struct fifonode *)pool_alloc(&fifos_pool, FALSE);
- pth_mutex_release(&clenvs_lock);
+ pthread_mutex_unlock(&clenvs_lock);
if (clenv) {
if (!(*CONF.DISPLAY_FILE) || fnode) {
free_clientenv(clenv);
} else {
if (fnode) {
- pth_mutex_acquire(&clenvs_lock, FALSE, NULL);
+ pthread_mutex_lock(&clenvs_lock);
pool_free((pnode_t *)fnode);
- pth_mutex_release(&clenvs_lock);
+ pthread_mutex_unlock(&clenvs_lock);
}
}
if (clenv->unode) {
bool zero = FALSE;
- pth_mutex_acquire(&lusers_lock, FALSE, NULL);
- pth_mutex_acquire(&clenv->unode->lock, FALSE, NULL);
+ pthread_mutex_lock(&lusers_lock);
+ pthread_mutex_lock(&clenv->unode->lock);
if ((--clenv->unode->logins) < 1)
zero = TRUE;
- pth_mutex_release(&clenv->unode->lock);
+ pthread_mutex_unlock(&clenv->unode->lock);
if (zero) {
hashtable_unlink(&lusers_table, (hashnode_t *)clenv->unode);
- /* May be required by some POSIX thread implementations */
- /* pth_mutex_destroy(&clenv->unode->lock); */
+ pthread_mutex_destroy(&clenv->unode->lock);
/* We now need to either update the cache entry for this home
* directory size or to create a new entry for it, but only if
*/
len = mm_strlen(home) + 1;
t = time(NULL);
- pth_mutex_acquire("a_lock, FALSE, NULL);
+ pthread_mutex_lock("a_lock);
if ((qnod = (quotanode *)hashtable_lookup("a_table, home,
len)) != NULL) {
qnod->homesize = clenv->unode->homesize;
qnod->dir, len, FALSE);
}
}
- pth_mutex_release("a_lock);
+ pthread_mutex_unlock("a_lock);
}
pool_free((pnode_t *)clenv->unode);
}
- pth_mutex_release(&lusers_lock);
+ pthread_mutex_unlock(&lusers_lock);
}
- pth_mutex_acquire(&clenvs_lock, FALSE, NULL);
- if (fnode != NULL) pool_free((pnode_t *)fnode);
- if (home != NULL) mmstrfree(home);
+ pthread_mutex_lock(&clenvs_lock);
+ if (fnode != NULL)
+ pool_free((pnode_t *)fnode);
+ if (home != NULL)
+ mmstrfree(home);
pool_free((pnode_t *)clenv);
- pth_mutex_release(&clenvs_lock);
+ pthread_mutex_unlock(&clenvs_lock);
return (NULL);
}
static bool
start_transfer_thread(clientenv *clenv)
{
- /* I do not understand why libpth doesn't provide a function to set the
- * reply port, or to init the message structure. Moreover, it theoretically
- * shouldn't need the message size if it simply swaps messages around port
- * linked list FIFOs via the message node. This currently seems the best
- * way to go about it
- */
- clenv->tmsg.msg.m_size = sizeof(transfermsg);
- clenv->tmsg.msg.m_replyport = clenv->rport;
+ pthread_msg_init(&clenv->tmsg.msg, &clenv->rport);
/*
- if ((clenv->tthread = pth_spawn(tthreadattr, (void *)transferthread,
- clenv))) {
- if (transfer_request(REQ_NONE, TRUE, clenv)) return (TRUE);
- pth_join(clenv->tthread, NULL);
+ if (pthread_create(&clenv->tthread, &tthreadattr, (void *)transferthread,
+ clenv) == 0) {
+ if (transfer_request(REQ_NONE, TRUE, clenv))
+ return (TRUE);
+ pthread_join(&clenv->tthread, NULL);
}
*/
- thread_object_call(NULL, transferthread, clenv);
+ if (pthread_object_call(NULL, transferthread, clenv) != 0)
+ DEBUG_PRINTF("start_transfer_thread", "pthread_object_call()");
+
if (transfer_request(REQ_NONE, TRUE, clenv))
return TRUE;
+ else
+ DEBUG_PRINTF("start_transfer_thread", "transfer_request(REQ_NONE)");
return (FALSE);
}
static void
stop_transfer_thread(clientenv *clenv)
{
- if (clenv->sport) {
+ if (pthread_port_valid(&clenv->sport)) {
/* Send a quit request and wait for reply */
if (!transfer_request(REQ_QUIT, TRUE, clenv))
DEBUG_PRINTF("stop_transfer_thread",
* Join thread as we want to make sure it exits before we do.
*/
/*
- pth_join(clenv->tthread, NULL);
+ pthread_join(&clenv->tthread, NULL);
*/
}
+ if (pthread_msg_valid(&clenv->tmsg.msg))
+ pthread_msg_destroy(&clenv->tmsg.msg);
}
{
int res = FALSE;
transfermsg *msg = &(clenv->tmsg);
- pth_event_t ring;
if (req != REQ_NONE) {
msg->request = req;
msg->result = FALSE;
- pth_msgport_put(clenv->sport, (pth_message_t *)msg);
+ pthread_msg_put(&clenv->sport, (pthread_msg_t *)msg);
}
if (waitreply) {
- ring = pth_event(PTH_EVENT_TIME, pth_timeout(120, 0));
- pth_event_concat(clenv->rring, ring, NULL);
for (;;) {
- if ((pth_wait(clenv->rring))) {
- if ((pth_event_occurred(ring))) {
- /* Timeout occured waiting for reply */
- res = FALSE;
- break;
- }
- if ((pth_event_occurred(clenv->rring))) {
- if ((msg = (transfermsg *)pth_msgport_get(clenv->rport))) {
- res = msg->result;
+ if ((msg = (transfermsg *)pthread_msg_get(&clenv->rport))
+ == NULL) {
+ struct timeval tv;
+ struct timespec ts, ts1;
+
+ ts1.tv_sec = 120;
+ ts1.tv_nsec = 0;
+ (void) gettimeofday(&tv, NULL);
+ TIMEVAL_TO_TIMESPEC(&tv, &ts);
+ timespecadd(&ts, &ts1, &ts);
+
+ if (pthread_ring_wait(&clenv->rring, &ts) == ETIMEDOUT) {
+ res = FALSE;
break;
- }
}
+ } else {
+ res = msg->result;
+ break;
}
}
- pth_event_isolate(clenv->rring);
- pth_event_free(ring, PTH_FREE_ALL);
} else
res = TRUE;
}
/* Open our reply port and start our file transfer thread */
- if ((clenv->rport = pth_msgport_create("XXX(NULL)"))) {
- clenv->rring = pth_event(PTH_EVENT_MSG, clenv->rport);
+ if (pthread_port_init(&clenv->rport) == 0 &&
+ pthread_ring_init(&clenv->rring) == 0 &&
+ pthread_port_set_ring(&clenv->rport, &clenv->rring) == 0) {
/* Init our clientenv as required */
clenv->fdb = fdb;
} else
DEBUG_PRINTF("handleclient", "start_transfer_thread()");
- pth_event_free(clenv->rring, PTH_FREE_ALL);
- while ((pth_msgport_get(clenv->rport)) != NULL) ;
- pth_msgport_destroy(clenv->rport);
+ while (pthread_msg_get(&clenv->rport) != NULL) ;
} else
- DEBUG_PRINTF("handleclient", "pth_msgport_create()");
+ DEBUG_PRINTF("handleclient", "pthread_port_init()");
+
+ if (pthread_port_valid(&clenv->rport))
+ pthread_port_destroy(&clenv->rport);
+ if (pthread_ring_valid(&clenv->rring))
+ pthread_ring_destroy(&clenv->rring);
control_in = FDBBYTESR(fdb);
control_out = FDBBYTESW(fdb);
p = CONF.PASV_RANGE_MIN;
}
if (err == -1)
- pth_sleep(1);
+ pthread_sleep(1);
}
if (err != -1) {
/* Successful bind(2) */
* other transfer requests when one is ongoing already.
*/
static void
-transferthread(struct thread_object *obj, void *args)
+transferthread(pthread_object_t *obj, void *args)
{
clientenv *clenv = args;
fdbuf *fdb;
if ((fdb = fdbopen(&gfdf, &fdbc, -1, FDB_BUFSIZE, FDB_BUFSIZE, 1024, 1024,
CONF.DATA_TIMEOUT * 1000, CONF.DATA_TIMEOUT * 1000,
FALSE)) != NULL) {
- if ((clenv->sport = pth_msgport_create("XXX(NULL)"))) {
+ if (pthread_port_init(&clenv->sport) == 0) {
/* Reply that we started thread properly */
ok = TRUE;
clenv->tmsg.result = TRUE;
- pth_msgport_put(clenv->rport, (pth_message_t *)&(clenv->tmsg));
+ pthread_msg_put(&clenv->rport, (pthread_msg_t *)&(clenv->tmsg));
/* Process our tasks */
transferthread_main(clenv, fdb);
- while ((msg = (transfermsg *)pth_msgport_get(clenv->sport))
+ while ((msg = (transfermsg *)pthread_msg_get(&clenv->sport))
!= NULL) {
msg->result = FALSE;
- pth_msgport_reply((pth_message_t *)msg);
+ pthread_msg_reply((pthread_msg_t *)msg);
}
- pth_msgport_destroy(clenv->sport);
+ pthread_port_destroy(&clenv->sport);
} else
DEBUG_PRINTF("transferthread", "pth_msgport_create()");
fdbclose(fdb);
if (!ok) {
/* Reply that thread couldn't be setup properly */
clenv->tmsg.result = FALSE;
- pth_msgport_put(clenv->rport, (pth_message_t *)&(clenv->tmsg));
+ pthread_msg_put(&clenv->rport, (pthread_msg_t *)&(clenv->tmsg));
}
/*
- pth_exit(NULL);
+ pthread_exit(NULL);
*/
}
static void
transferthread_main(clientenv *clenv, fdbuf *fdb)
{
- int state = TTSTATE_INITIAL, l, reason;
+ int state = TTSTATE_INITIAL, l, reason, err;
register char *from, *to;
char ipaddr[20], buffer[T_BUFSIZE], *from2;
- pth_event_t ring, ring1;
+ pthread_ring_t ring;
transfermsg *msg;
struct sockaddr_in server, client, addr;
socklen_t addrl;
fdbparam_set(fdb, -1, FDBP_FD);
/* Transfer device entry setup */
- ring = pth_event(PTH_EVENT_MSG, clenv->sport);
+ pthread_ring_init(&ring);
+ pthread_port_set_ring(&clenv->sport, &ring);
/* Main device loop */
while (state != TTSTATE_DONE) {
/* State main loop */
while (state == TTSTATE_INITIAL) {
- if ((pth_wait(ring))) {
- if ((pth_event_occurred(ring))) {
- while ((msg = (transfermsg *)pth_msgport_get(
- clenv->sport)) != NULL) {
- /* Process request and reply */
- switch (msg->request) {
- case REQ_QUIT:
- msg->result = TRUE;
- state = TTSTATE_DONE;
- break;
- case REQ_ABORT:
- /* As we handle PASV port, do this */
- if (fdpasv != -1) {
- close(fdpasv);
- fdpasv = -1;
+ if (pthread_port_pending(&clenv->sport) < 1)
+ pthread_ring_wait(&ring, NULL);
+ while ((msg = (transfermsg *)pthread_msg_get(&clenv->sport))
+ != NULL) {
+ /* Process request and reply */
+ switch (msg->request) {
+ case REQ_QUIT:
+ msg->result = TRUE;
+ state = TTSTATE_DONE;
+ break;
+ case REQ_ABORT:
+ /* As we handle PASV port, do this */
+ if (fdpasv != -1) {
+ close(fdpasv);
+ fdpasv = -1;
+ }
+ port = 0;
+ passive = ongoing = FALSE;
+ download = TRUE;
+ list = 0;
+ *path = '\0';
+ msg->result = TRUE;
+ break;
+ case REQ_STATUS:
+ msg->port = (u_int16_t)port;
+ msg->passive = passive;
+ msg->ongoing = ongoing;
+ msg->download = download;
+ msg->list = list;
+ msg->rbytes = FDBBYTESR(fdb);
+ msg->wbytes = FDBBYTESW(fdb);
+ msg->dlfiles = dlfiles;
+ msg->ulfiles = ulfiles;
+ msg->result = TRUE;
+ break;
+ case REQ_NEWPORT:
+ port = 0;
+ if (fd != -1) {
+ fdbflushw(fdb);
+ shutdown(fd, SHUT_RDWR);
+ close(fd);
+ fd = -1;
+ fdbparam_set(fdb, -1, FDBP_FD);
+ }
+ /* As we handle PASV port, do this */
+ if (fdpasv != -1) {
+ close(fdpasv);
+ fdpasv = -1;
+ }
+ if (msg->passive) {
+ /* PASV request, we need to open a high
+ * port and start listening for a
+ * connection on it
+ */
+ msg->result = FALSE; /* Default */
+ if (pasv_bind(clenv, &server, &fdpasv,
+ &port)) {
+ msg->port = (u_int16_t)port;
+ passive = msg->result = TRUE;
+ break;
}
- port = 0;
- passive = ongoing = FALSE;
- download = TRUE;
- list = 0;
- *path = '\0';
- msg->result = TRUE;
- break;
- case REQ_STATUS:
- msg->port = (u_int16_t)port;
- msg->passive = passive;
- msg->ongoing = ongoing;
- msg->download = download;
- msg->list = list;
- msg->rbytes = FDBBYTESR(fdb);
- msg->wbytes = FDBBYTESW(fdb);
- msg->dlfiles = dlfiles;
- msg->ulfiles = ulfiles;
+ } else {
+ /* Normal PORT, just remember necessary
+ * things. The caller should have
+ * performed address and port sanity
+ * checking already.
+ */
+ port = (int)msg->port;
+ passive = FALSE;
msg->result = TRUE;
- break;
- case REQ_NEWPORT:
- port = 0;
- if (fd != -1) {
- fdbflushw(fdb);
- shutdown(fd, SHUT_RDWR);
- close(fd);
- fd = -1;
- fdbparam_set(fdb, -1, FDBP_FD);
- }
- /* As we handle PASV port, do this */
- if (fdpasv != -1) {
- close(fdpasv);
- fdpasv = -1;
- }
- if (msg->passive) {
- /* PASV request, we need to open a high
- * port and start listening for a
- * connection on it
- */
- msg->result = FALSE; /* Default */
- if (pasv_bind(clenv, &server, &fdpasv,
- &port)) {
- msg->port = (u_int16_t)port;
- passive = msg->result = TRUE;
- break;
- }
- } else {
- /* Normal PORT, just remember necessary
- * things. The caller should have
- * performed address and port sanity
- * checking already.
- */
- port = (int)msg->port;
- passive = FALSE;
+ }
+ break;
+ case REQ_TRANSFER:
+ msg->result = FALSE; /* Default */
+ if (port != 0) {
+ fdbparam_set(fdb, msg->rrate * 1024,
+ FDBP_RRATE);
+ fdbparam_set(fdb, msg->wrate * 1024,
+ FDBP_WRATE);
+ if (msg->path) mm_strncpy(path, msg->path,
+ MMPATH_MAX - 1);
+ else *path = '\0';
+ if (msg->list) {
+ list = msg->list;
msg->result = TRUE;
- }
- break;
- case REQ_TRANSFER:
- msg->result = FALSE; /* Default */
- if (port != 0) {
- fdbparam_set(fdb, msg->rrate * 1024,
- FDBP_RRATE);
- fdbparam_set(fdb, msg->wrate * 1024,
- FDBP_WRATE);
- if (msg->path) mm_strncpy(path, msg->path,
- MMPATH_MAX - 1);
- else *path = '\0';
- if (msg->list) {
- list = msg->list;
+ } else {
+ if (msg->file != -1) {
+ file = msg->file;
+ download = msg->download;
msg->result = TRUE;
- } else {
- if (msg->file != -1) {
- file = msg->file;
- download = msg->download;
- msg->result = TRUE;
- }
}
- if (msg->result)
- state = TTSTATE_CONNECT;
- } else
- msg->result = FALSE;
- break;
- default:
+ }
+ if (msg->result)
+ state = TTSTATE_CONNECT;
+ } else
msg->result = FALSE;
- }
- pth_msgport_reply((pth_message_t *)msg);
- }
+ break;
+ default:
+ msg->result = FALSE;
}
+ pthread_msg_reply((pthread_msg_t *)msg);
}
}
/* State entry setup */
ongoing = TRUE;
- /* Setup our connect timeout event */
- ring1 = pth_event(PTH_EVENT_TIME, pth_timeout(CONF.DATA_TIMEOUT,
- 0));
- pth_event_concat(ring, ring1, NULL);
/* State main loop */
while (state == TTSTATE_CONNECT) {
if (passive) {
- /* Accept connection from client */
+ /*
+ * Accept connection from client.
+ * On timeout, ETIMEDOUT is returned, on interrupting
+ * message, ECANCELED is. Other errror codes could be
+ * returned, or the new file descriptor.
+ */
addrl = sizeof(struct sockaddr);
- fd = pth_accept_ev(fdpasv, (struct sockaddr *)&addr,
- &addrl, ring);
+ if ((err = pthread_accept_ring(fdpasv,
+ (struct sockaddr *)&addr,
+ &addrl, CONF.DATA_TIMEOUT * 1000,
+ &ring)) > -1) {
+ fd = err;
+ err = 0;
+ } else {
+ fd = -1;
+ err = errno;
+ }
} else {
/* Start connection to the client */
if ((fd = socket(AF_INET, SOCK_STREAM, 0)) != -1) {
client.sin_family = AF_INET;
client.sin_addr.s_addr = clenv->cipaddr;
client.sin_port = htons(port);
- /* Finally attempt connection */
- if ((pth_connect_ev(fd, (struct sockaddr *)&client,
+ /*
+ * Finally attempt connection.
+ * On timeout, ETIMEDOUT is returned, on interrupting
+ * message, ECANCELED is. Other error codes could be
+ * returned, or 0 on success. In the case of any of
+ * the two special error codes we just described, the
+ * connection request is still pending in the kernel.
+ * We can continue calling the following function
+ * until we get connected, which can return EISCONN if
+ * it already was connected, or we could monitor the
+ * status of the connection using getsockopt(2) with
+ * SOL_SOCKET level and SO_ERROR option. If we want
+ * to abort the in-progress connection, we need to use
+ * close(2).
+ */
+ /*
+ * XXX Hmm we should loop here! Not create new sockets
+ * again in the loop! Unless we wanted to make
+ * pthread_connect_ring() automatically cancel any
+ * connection closing the descriptor upon ETIMEDOUT or
+ * ECANCELED return codes, in which case we could
+ * continue the current behavior.
+ */
+ if ((pthread_connect_ring(fd,
+ (struct sockaddr *)&client,
sizeof(struct sockaddr_in),
- ring)) == -1) {
+ CONF.DATA_TIMEOUT * 1000,
+ &ring)) != 0 && errno != ETIMEDOUT &&
+ errno != ECANCELED) {
mmsyslog(1, LOGLEVEL,
"%08X PORT data connection failiure",
clenv->id);
REGISTER_ERROR();
state = TTSTATE_INITIAL;
break;
- }
+ } else if (errno == ETIMEDOUT || errno == ECANCELED) {
+ /* XXX For now */
+ (void) close(fd);
+ fd = -1;
+ err = ETIMEDOUT;
+ } else
+ err = 0;
} else {
DEBUG_PRINTF("transferthread_main-2", "socket()");
reply(clenv->fdb, 425, FALSE,
}
}
/* Verify if connection timeout occured */
- if ((pth_event_occurred(ring1))) {
+ if (err == ETIMEDOUT) {
mmsyslog(1, LOGLEVEL, "%08X Data connection timeout",
clenv->id);
reply(clenv->fdb, 425, FALSE,
state = TTSTATE_INITIAL;
break;
}
- /* Check for any request events and process them if any */
- if ((pth_event_occurred(ring))) {
- while ((msg = (transfermsg *)pth_msgport_get(
- clenv->sport)) != NULL) {
- /* Process request and reply */
- switch (msg->request) {
- case REQ_QUIT:
- msg->result = TRUE;
- state = TTSTATE_DONE;
- break;
- case REQ_ABORT:
- msg->result = TRUE;
- state = TTSTATE_INITIAL;
- break;
- case REQ_STATUS:
- msg->port = (u_int16_t)port;
- msg->passive = passive;
- msg->ongoing = ongoing;
- msg->download = download;
- msg->list = list;
- msg->rbytes = FDBBYTESR(fdb);
- msg->wbytes = FDBBYTESW(fdb);
- msg->dlfiles = dlfiles;
- msg->ulfiles = ulfiles;
- msg->result = TRUE;
- break;
- default:
- msg->result = FALSE;
- }
- pth_msgport_reply((pth_message_t *)msg);
+ /*
+ * Check for any request events and process them if any.
+ * ECANCELED above can also cause this code execution to be
+ * triggered fast.
+ */
+ while ((msg = (transfermsg *)pthread_msg_get(&clenv->sport))
+ != NULL) {
+ /* Process request and reply */
+ switch (msg->request) {
+ case REQ_QUIT:
+ msg->result = TRUE;
+ state = TTSTATE_DONE;
+ break;
+ case REQ_ABORT:
+ msg->result = TRUE;
+ state = TTSTATE_INITIAL;
+ break;
+ case REQ_STATUS:
+ msg->port = (u_int16_t)port;
+ msg->passive = passive;
+ msg->ongoing = ongoing;
+ msg->download = download;
+ msg->list = list;
+ msg->rbytes = FDBBYTESR(fdb);
+ msg->wbytes = FDBBYTESW(fdb);
+ msg->dlfiles = dlfiles;
+ msg->ulfiles = ulfiles;
+ msg->result = TRUE;
+ break;
+ default:
+ msg->result = FALSE;
}
+ pthread_msg_reply((pthread_msg_t *)msg);
}
if (state == TTSTATE_CONNECT) {
/* Verify if connection was established */
}
/* State exit cleanup */
- pth_event_isolate(ring);
- pth_event_free(ring1, PTH_FREE_ALL);
} else if (state == TTSTATE_TRANSFER) { /* TRANSFER STATE */
TYPE_IMAGE ? fdbrcleartosend(fdb) :
fdbcleartosend(fdb))) {
/* Client ready to be written some bytes to */
- if ((l = pth_read(file, buffer, T_BUFSIZE)) > 0) {
+ if ((l = read(file, buffer, T_BUFSIZE)) > 0) {
if (clenv->type == TYPE_ASCII) {
/* ASCII xfer, perform \n to \r\n
* conversion
break;
}
/* Verify if we received any important events */
- if ((pth_event_occurred(ring))) {
- while ((msg = (transfermsg *)pth_msgport_get(
- clenv->sport)) != NULL) {
- /* Process request and reply */
- switch (msg->request) {
- case REQ_QUIT:
- msg->result = TRUE;
- state = TTSTATE_DONE;
- cont = FALSE;
- break;
- case REQ_ABORT:
- msg->result = TRUE;
- state = TTSTATE_INITIAL;
- cont = FALSE;
- reason = TR_ABORTED;
- break;
- case REQ_STATUS:
- msg->port = (u_int16_t)port;
- msg->passive = passive;
- msg->ongoing = ongoing;
- msg->download = download;
- msg->list = list;
- msg->rbytes = FDBBYTESR(fdb);
- msg->wbytes = FDBBYTESW(fdb);
- msg->dlfiles = dlfiles;
- msg->ulfiles = ulfiles;
- msg->result = TRUE;
- break;
- default:
- msg->result = FALSE;
- }
- pth_msgport_reply((pth_message_t *)msg);
+ while ((msg = (transfermsg *)pthread_msg_get(
+ &clenv->sport)) != NULL) {
+ /* Process request and reply */
+ switch (msg->request) {
+ case REQ_QUIT:
+ msg->result = TRUE;
+ state = TTSTATE_DONE;
+ cont = FALSE;
+ break;
+ case REQ_ABORT:
+ msg->result = TRUE;
+ state = TTSTATE_INITIAL;
+ cont = FALSE;
+ reason = TR_ABORTED;
+ break;
+ case REQ_STATUS:
+ msg->port = (u_int16_t)port;
+ msg->passive = passive;
+ msg->ongoing = ongoing;
+ msg->download = download;
+ msg->list = list;
+ msg->rbytes = FDBBYTESR(fdb);
+ msg->wbytes = FDBBYTESW(fdb);
+ msg->dlfiles = dlfiles;
+ msg->ulfiles = ulfiles;
+ msg->result = TRUE;
+ break;
+ default:
+ msg->result = FALSE;
}
+ pthread_msg_reply((pthread_msg_t *)msg);
}
}
if (*from == '\r') {
if (treesize_edit(clenv,
from - from2)) {
- if ((l = pth_write(file,
- from2, from - from2))
- != from - from2) {
+ if ((l = write(file, from2,
+ from - from2)) !=
+ from - from2) {
mmsyslog(0, LOGLEVEL,
"%08X Error "
"writing to "
}
if (l && from2 < to) {
if (treesize_edit(clenv, to - from2)) {
- if ((l = pth_write(file, from2,
- to - from2))
- != to - from2) {
+ if ((l = write(file, from2,
+ to - from2)) != to - from2) {
mmsyslog(0, LOGLEVEL,
"%08X Error writing "
"to disk!",
/* IMAGE/binary transfer, write data as-is
*/
if (treesize_edit(clenv, l)) {
- if ((pth_write(file, buffer, l)) !=
- l) {
+ if ((write(file, buffer, l)) != l) {
mmsyslog(0, LOGLEVEL,
"%08X Error writing to "
"disk!", clenv->id);
break;
}
/* Verify if we received any important events */
- if ((pth_event_occurred(ring))) {
- while ((msg = (transfermsg *)pth_msgport_get(
- clenv->sport)) != NULL) {
- /* Process request and reply */
- switch (msg->request) {
- case REQ_QUIT:
- msg->result = TRUE;
- state = TTSTATE_DONE;
- cont = FALSE;
- break;
- case REQ_ABORT:
- msg->result = TRUE;
- state = TTSTATE_INITIAL;
- cont = FALSE;
- reason = TR_ABORTED;
- break;
- case REQ_STATUS:
- msg->port = (u_int16_t)port;
- msg->passive = passive;
- msg->ongoing = ongoing;
- msg->download = download;
- msg->list = list;
- msg->rbytes = FDBBYTESR(fdb);
- msg->wbytes = FDBBYTESW(fdb);
- msg->dlfiles = dlfiles;
- msg->ulfiles = ulfiles;
- msg->result = TRUE;
- break;
- default:
- msg->result = FALSE;
- }
- pth_msgport_reply((pth_message_t *)msg);
+ while ((msg = (transfermsg *)pthread_msg_get(
+ &clenv->sport)) != NULL) {
+ /* Process request and reply */
+ switch (msg->request) {
+ case REQ_QUIT:
+ msg->result = TRUE;
+ state = TTSTATE_DONE;
+ cont = FALSE;
+ break;
+ case REQ_ABORT:
+ msg->result = TRUE;
+ state = TTSTATE_INITIAL;
+ cont = FALSE;
+ reason = TR_ABORTED;
+ break;
+ case REQ_STATUS:
+ msg->port = (u_int16_t)port;
+ msg->passive = passive;
+ msg->ongoing = ongoing;
+ msg->download = download;
+ msg->list = list;
+ msg->rbytes = FDBBYTESR(fdb);
+ msg->wbytes = FDBBYTESW(fdb);
+ msg->dlfiles = dlfiles;
+ msg->ulfiles = ulfiles;
+ msg->result = TRUE;
+ break;
+ default:
+ msg->result = FALSE;
}
+ pthread_msg_reply((pthread_msg_t *)msg);
}
}
}
}
if (fdpasv != -1) close(fdpasv);
- pth_event_free(ring, PTH_FREE_ALL);
+ pthread_ring_destroy(&ring);
}
/* mmfd library thread support functions */
+static pthread_mutexattr_t thread_ma;
+
+static void
+thread_init(void)
+{
+ (void) pthread_mutexattr_init(&thread_ma);
+ (void) pthread_mutexattr_settype(&thread_ma, PTHREAD_MUTEX_RECURSIVE);
+}
static void *
-_pth_mutex_create(void)
+thread_mutex_create(void)
{
struct mutexnode *mnod;
- pth_mutex_acquire(&mutexes_lock, FALSE, NULL);
+ pthread_mutex_lock(&mutexes_lock);
mnod = (struct mutexnode *)pool_alloc(&mutexes_pool, FALSE);
- pth_mutex_release(&mutexes_lock);
+ pthread_mutex_unlock(&mutexes_lock);
- if (mnod)
- pth_mutex_init(&mnod->mutex);
+ if (mnod != NULL)
+ pthread_mutex_init(&mnod->mutex, &thread_ma);
return ((void *)mnod);
}
static void *
-_pth_mutex_destroy(void *mtx)
+thread_mutex_destroy(void *mtx)
{
- /* struct mutexnode *mnod = mtx; */
+ struct mutexnode *mnod = mtx;
- /* pth_mutex_destroy(&mnod->mutex); */
- pth_mutex_acquire(&mutexes_lock, FALSE, NULL);
+ pthread_mutex_destroy(&mnod->mutex);
+
+ pthread_mutex_lock(&mutexes_lock);
pool_free(mtx);
- pth_mutex_release(&mutexes_lock);
+ pthread_mutex_unlock(&mutexes_lock);
return (NULL);
}
static void
-_pth_mutex_lock(void *mtx)
+thread_mutex_lock(void *mtx)
{
struct mutexnode *mnod = mtx;
- pth_mutex_acquire(&mnod->mutex, FALSE, NULL);
+ pthread_mutex_lock(&mnod->mutex);
}
static void
-_pth_mutex_unlock(void *mtx)
+thread_mutex_unlock(void *mtx)
{
struct mutexnode *mnod = mtx;
- pth_mutex_release(&mnod->mutex);
-}
-
-
-static void
-_pth_thread_yield(void)
-{
- pth_yield(NULL);
+ pthread_mutex_unlock(&mnod->mutex);
}
static bool
-_pth_eintr(void)
+thread_eintr(void)
{
if (errno == EINTR)
return TRUE;
found = FALSE;
size = 0;
len = mm_strlen(path) + 1;
- pth_mutex_acquire("a_lock, FALSE, NULL);
+ pthread_mutex_lock("a_lock);
if ((qnod = (quotanode *)hashtable_lookup("a_table, path, len))
!= NULL) {
struct stat st;
}
}
}
- pth_mutex_release("a_lock);
+ pthread_mutex_unlock("a_lock);
if (found)
return (size);
-/* $Id: mmserver.c,v 1.34 2005/02/20 01:18:01 mmondor Exp $ */
+/* $Id: mmserver.c,v 1.34.2.1 2005/11/16 01:00:58 mmondor Exp $ */
/*
* Copyright (C) 2000-2004, Matthew Mondor
#include <stdlib.h>
#include <unistd.h>
#include <stdio.h>
+#include <string.h>
+#include <errno.h>
#include <netdb.h>
#include <netinet/in.h>
#include <syslog.h>
-#include <pth.h>
+#include <pthread.h>
#include <signal.h>
#include <sys/poll.h>
#include <time.h>
#include <mmreadcfg.h>
#include <mmlimitrate.h>
#include <mmlog.h>
-#include <mm_pth_pool.h>
+
+#include <mm_pthread_pool.h>
+#include <mm_pthread_msg.h>
+#include <mm_pthread_poll.h>
+#include <mm_pthread_sleep.h>
MMCOPYRIGHT("@(#) Copyright (c) 2000-2004\n\
\tMatthew Mondor. All rights reserved.\n");
-MMRCSID("$Id: mmserver.c,v 1.34 2005/02/20 01:18:01 mmondor Exp $");
+MMRCSID("$Id: mmserver.c,v 1.34.2.1 2005/11/16 01:00:58 mmondor Exp $");
static void sighandler(int);
static struct iface * parse_ifaces(char *, char *);
-static void phandleclient(struct thread_object *, void *);
+static void phandleclient(pthread_object_t *, void *);
static void writepidfile(const char *);
static void * async_thread(void *);
static struct async_clenv * async_open_clenv(void);
static int (*handleclientfunc)(unsigned long, int,
clientlistnode *, struct iface *,
struct async_clenv *);
-static pth_mutex_t ctable_lock, ppool_lock;
+static pthread_mutex_t ctable_lock, ppool_lock;
static pool_t cpool, ppool;
static hashtable_t ctable;
static struct async_env *async = NULL;
/* MAIN DAEMON PROGRAM */
-/* Before calling this function, the async_init() and async_init_pth()
+/* Before calling this function, the async_init() and async_init_pthread()
* functions should have been called. Before any async_*() or tcp_server()
* calls the process is expected to have called make_daemon().
*/
struct ifacendx *fdsi;
struct iface *ifaces, *tif;
unsigned long id;
- pth_t clnode_thread;
- pth_attr_t threadattr, clnode_threadattr;
- int tcp_proto;
+ pthread_t clnode_thread;
+ pthread_attr_t threadattr, clnode_threadattr;
+ int tcp_proto, clnode_thread_started, err;
sinaddr = (struct sockaddr_in *)&addr;
handleclientfunc = handleclient1;
fds = NULL;
fdsi = NULL;
ifaces = NULL;
+ clnode_thread_started = -1;
- /* Pth related */
- threadattr = pth_attr_new();
- pth_attr_set(threadattr, PTH_ATTR_JOINABLE, FALSE);
- clnode_threadattr = pth_attr_new();
- pth_attr_set(clnode_threadattr, PTH_ATTR_JOINABLE, TRUE);
- pth_mutex_init(&ctable_lock);
- pth_mutex_init(&ppool_lock);
+ /* Pthread related */
+ pthread_attr_init(&threadattr);
+ pthread_attr_setdetachstate(&threadattr, 1);
+ pthread_attr_init(&clnode_threadattr);
+ pthread_attr_setdetachstate(&clnode_threadattr, 0);
+ pthread_mutex_init(&ctable_lock, NULL);
+ pthread_mutex_init(&ppool_lock, NULL);
/* Used by resolve_hostname() */
if (!mmstrinit(malloc, free, 16384)) {
sizeof(phandleinfo), 16384 / sizeof(phandleinfo), 0, 0) ||
!hashtable_init(&ctable, "ctable", maxips, 1, malloc, free,
clnode_keycmp, clnode_keyhash, FALSE) ||
- ((clnode_thread = pth_spawn(clnode_threadattr,
- clnode_expire_thread, &rateper)) == NULL)) {
+ ((clnode_thread_started = pthread_create(&clnode_thread,
+ &clnode_threadattr, clnode_expire_thread, &rateper))) != 0) {
syslog(LOG_NOTICE, "tcp_server() - Out of memory");
closelog();
mmstrexit();
exit(-1);
}
- if (thread_object_init() == -1) {
- syslog(LOG_NOTICE, "tcp_server() - thread_object_init()");
+ if ((err = pthread_object_init(8)) != 0) {
+ syslog(LOG_NOTICE, "tcp_server() - pthread_object_init() - %s",
+ strerror(err));
exit(-1);
}
if (ret) {
free(ifaces);
mmstrexit();
- if (clnode_thread != NULL) {
- pth_abort(clnode_thread);
- pth_join(clnode_thread, NULL);
+ if (clnode_thread_started == 0) {
+ pthread_cancel(clnode_thread);
+ pthread_join(clnode_thread, NULL);
}
if (HASHTABLE_VALID(&ctable))
hashtable_destroy(&ctable, FALSE);
/* Start answering and dispatching connections */
syslog(LOG_NOTICE, "Started for uid: %d", uid);
for (;;) {
- if ((nifaces2 = pth_poll(fds, nifaces, -1)) > 0) {
+ if ((nifaces2 = poll(fds, nifaces, -1)) > 0) {
/* Loop but once only, and only for long as nifaces2 times.
* Use our fast index to reference to the interface structure.
*/
if (fds[i].revents & POLLIN) {
nifaces2--;
addrl = sizeof(struct sockaddr);
- if ((msgsock = pth_accept(fds[i].fd, &addr, &addrl))
- != -1) {
+ if ((msgsock = accept(fds[i].fd, &addr, &addrl)) != -1) {
/* Make sure that we respect connection and rate
* limits.
*/
ok = FALSE;
reason = MMS_NORMAL;
- pth_mutex_acquire(&ctable_lock, FALSE, NULL);
+ pthread_mutex_lock(&ctable_lock);
if ((clnode = (clientlistnode *)hashtable_lookup(
&ctable, &sinaddr->sin_addr.s_addr,
sizeof(u_int32_t))) == NULL) {
} else
reason = MMS_CONPERADDRESS_EXCEEDED;
}
- pth_mutex_release(&ctable_lock);
+ pthread_mutex_unlock(&ctable_lock);
if (ok) {
phandleinfo *phi;
/* Dispatch client to a thread */
- pth_mutex_acquire(&ppool_lock, FALSE, NULL);
+ pthread_mutex_lock(&ppool_lock);
phi = (phandleinfo *)pool_alloc(&ppool, FALSE);
- pth_mutex_release(&ppool_lock);
+ pthread_mutex_unlock(&ppool_lock);
if (phi != NULL) {
id++;
phi->id = id;
phi->clnode = clnode;
/* Indexed reference */
phi->iface = fdsi[i].iface;
- if (thread_object_call(NULL, phandleclient,
- phi) != -1)
+ if (pthread_object_call(NULL, phandleclient,
+ phi) == 0)
clnode->connections++;
else {
if (phi != NULL) {
- pth_mutex_acquire(&ppool_lock, FALSE,
- NULL);
+ pthread_mutex_lock(&ppool_lock);
pool_free((pnode_t *)phi);
- pth_mutex_release(&ppool_lock);
+ pthread_mutex_unlock(&ppool_lock);
}
syslog(LOG_NOTICE, "tcp_server() - "
"Error launching thread");
/* Close down connection with client
* immediately, sending message.
*/
- (void) pth_write(msgsock, message, msglen);
+ (void) write(msgsock, message, msglen);
shutdown(msgsock, 2);
close(msgsock);
mm_strcpy(ipaddr, "0.0.0.0");
syslog(LOG_NOTICE, "tcp_server() - [%s] - %s",
ipaddr, MMS_RSTRING(reason));
}
-
- pth_mutex_release(&ctable_lock);
-
} else
- syslog(LOG_NOTICE, "tcp_server() - pth_accept()");
+ syslog(LOG_NOTICE, "tcp_server() - accept()");
}
}
}
free(ifaces);
mmstrexit();
- pth_abort(clnode_thread);
- pth_join(clnode_thread, NULL);
+ pthread_cancel(clnode_thread);
+ pthread_join(clnode_thread, NULL);
hashtable_destroy(&ctable, FALSE);
pool_destroy(&ppool);
pool_destroy(&cpool);
- /* Unrequired for Pth (non-existing even)
- * pth_mutex_destroy(&apool_lock);
- * pth_mutex_destroy(&ctable_lock);
- * pth_mutex_destroy(&ppool_lock);
- */
+ pthread_mutex_destroy(&ctable_lock);
+ pthread_mutex_destroy(&ppool_lock);
exit(ret);
}
/* SERVER (THREAD) */
-/* This is started by pth_spawn(), it calls the user handler function, which
- * only has to care about serving the client.
+/* This is started by pthread_create(), it calls the user handler function,
+ * which only has to care about serving the client.
*/
static void
-phandleclient(struct thread_object *obj, void *args)
+phandleclient(pthread_object_t *obj, void *args)
{
phandleinfo *phi;
int socket, ret;
socket = phi->socket;
clnode = phi->clnode;
iface = phi->iface;
- pth_mutex_acquire(&ppool_lock, FALSE, NULL);
+ pthread_mutex_lock(&ppool_lock);
pool_free((pnode_t *)phi);
- pth_mutex_release(&ppool_lock);
+ pthread_mutex_unlock(&ppool_lock);
ret = 0;
}
if (tmp) {
/* Lock the mutex for a very short moment */
- pth_mutex_acquire(&ctable_lock, FALSE, NULL);
+ pthread_mutex_lock(&ctable_lock);
/* Verify again for NULL to avoid a potential memory leak */
if(clnode->hostname == NULL)
clnode->hostname = tmp;
else
tmp = mmstrfree(tmp);
- pth_mutex_release(&ctable_lock);
+ pthread_mutex_unlock(&ctable_lock);
}
}
/* Decrease our connection/refcount counter, and let our ctable thread
* decide if/when to uncache our node.
*/
- pth_mutex_acquire(&ctable_lock, FALSE, NULL);
+ pthread_mutex_lock(&ctable_lock);
if (clnode->connections > 0)
clnode->connections--;
- pth_mutex_release(&ctable_lock);
+ pthread_mutex_unlock(&ctable_lock);
/* kthxbye :) */
}
/* Fatal error */
syslog(LOG_NOTICE, "async_thread() - calloc(%d)",
idx_size + pfd_size);
- pth_exit(NULL);
+ pthread_exit(NULL);
}
i = 0;
DLIST_FOREACH(freeprocs, proc) {
/* First check for new messages and queue them in our queue list if
* any, but only wait for them if all processes are free (we don't
- * expect any results). Special note: Pth will only notify event for
- * a port if it received a message while it was empty.
+ * expect any results).
*/
- if ((pth_msgport_pending(async->port))) {
- while ((omsg = (char *)pth_msgport_get(async->port)) != NULL)
+ if (pthread_port_pending(&async->port) > 0) {
+ while ((omsg = (char *)pthread_msg_get(&async->port)) != NULL)
DLIST_APPEND(&queue, (node_t *)(omsg - sizeof(pnode_t)));
- } else if (freeprocs->nodes == async->nprocs) {
- pth_wait(async->ring);
- while ((omsg = (char *)pth_msgport_get(async->port)) != NULL)
+ } else if (DLIST_NODES(freeprocs) == async->nprocs) {
+ /* XXX */ syslog(LOG_NOTICE, "* pthread_ring_wait() starting");
+ pthread_ring_wait(&async->ring, NULL);
+ /* XXX */ syslog(LOG_NOTICE, "* pthread_ring_wait() ended");
+ while ((omsg = (char *)pthread_msg_get(&async->port)) != NULL)
DLIST_APPEND(&queue, (node_t *)(omsg - sizeof(pnode_t)));
}
if (DEBUG_TRUE(m->func_id < async->nfuncs)) {
len = async->funcs[m->func_id].msg_len;
idx[p->sock].aclenv = m->aclenv;
- if ((pth_send(p->sock, m, len, 0)) == len)
+ /* XXX */ syslog(LOG_NOTICE, "send()ing");
+ if ((send(p->sock, m, len, 0)) == len)
DLIST_UNLINK(freeprocs, (node_t *)p);
else
syslog(LOG_NOTICE, "async_thread() - send(%d:%d)",
}
/* Wait for results from our async processes via their socket,
- * but be sure to break when new Pth messages are available.
+ * but be sure to break when new inter-thread messages are available.
* If new results are obtained, reply back to the caller thread.
*/
for (;;) {
int selr, i;
- if ((selr = pth_poll_ev(pfds, nfds, -1, async->ring)) > 0) {
+ /* XXX */ syslog(LOG_NOTICE, "* pthread_poll_ring() starting");
+ selr = pthread_poll_ring(pfds, nfds, -1, &async->ring);
+ /* XXX */ syslog(LOG_NOTICE, "* pthread_poll_ring() ended (selr = "
+ "%d (%s)", selr, strerror(errno));
+ if (selr == -1 && errno == ECANCELED)
+ break;
+ if (selr > 0) {
for (i = 0; selr > 0 && i < nfds; i++) {
if (pfds[i].revents & POLLERR) {
/* If this happens something is really wrong, exit */
/* Results, reply message back to client thread and
* return this process node in the free pool.
*/
- if ((len = pth_recv(fd, e->msg, async->msg_len,
+ if ((len = recv(fd, e->msg, async->msg_len,
MSG_WAITALL)) <
sizeof(struct async_msg))
syslog(LOG_NOTICE, "async_thread() - recv(%d:%d)",
fd, (int)async->msg_len);
- pth_msgport_reply(&(e->msg->msg));
+ /* XXX */syslog(LOG_NOTICE, "recv()ed reply");
+ pthread_msg_reply(&(e->msg->msg));
DLIST_APPEND(freeprocs, (node_t *)p);
selr--;
}
}
}
- if ((pth_event_occurred(async->ring)))
- /* Other requests we must queue immediately */
- break;
}
}
/* NOTREACHED */
- pth_exit(NULL);
+ pthread_exit(NULL);
return (NULL);
}
/* This function should be called at early application startup to setup
- * the asynchroneous pool of processes. async_init_pth() call may then
- * be called later on after pth_init() to initialize the client-side
+ * the asynchroneous pool of processes. async_init_pthread() call may then
+ * be called later on after pthread_init() to initialize the client-side
* asynchroneous context (required before using async_open_clenv()).
* This function is assumed to be called once only, and it's effects to
* only be discarded when the main process exits.
}
-/* This function is used by an application after pth_init() to initialize
+/* This function is used by an application after pthread_init() to initialize
* the asynchroneous thread device/server, which will dispatch requests to
* the asynchroneous pool of processes in a thread-safe manner (without
* blocking the whole process. async_init() should have been called first.
*/
bool
-async_init_pth(void)
+async_init_pthread(void)
{
size_t env_len;
bool res = FALSE;
if (DEBUG_TRUE(async)) {
+ pthread_poll_init();
+
res = TRUE;
- pth_mutex_init(&async->apool_lock);
+ pthread_mutex_init(&async->apool_lock, NULL);
/* Setup our fast-allocation pool for client-side async environments */
env_len = (size_t)OALIGN_CEIL(sizeof(struct async_clenv), long);
if (!pool_init(&async->apool, "asyncenv_pool", malloc, free,
aclenv_constructor, aclenv_destructor,
env_len + async->msg_len,
16384 / (env_len + async->msg_len), 0, 0)) {
- syslog(LOG_NOTICE, "async_init_pth() - pool_init(apool)");
+ syslog(LOG_NOTICE, "async_init_pthread() - pool_init(apool)");
res = FALSE;
}
if (res) {
- pth_attr_t attrs;
+ pthread_attr_t attrs;
+ pthread_t thread;
res = FALSE;
- if ((attrs = pth_attr_new()) != NULL) {
- /* Start async slave thread device */
- if ((async->port = pth_msgport_create("mms_async_server"))) {
- if ((async->ring = pth_event(PTH_EVENT_MSG,
- async->port))) {
- pth_attr_set(attrs, PTH_ATTR_JOINABLE, FALSE);
- if ((pth_spawn(attrs, async_thread, NULL)) != NULL)
- res = TRUE;
- pth_attr_destroy(attrs);
- } else
- syslog(LOG_NOTICE, "async_init_pth() - pth_event()");
+ pthread_attr_init(&attrs);
+ /* Start async slave thread device */
+ if (pthread_port_init(&async->port) == 0) {
+ if (pthread_ring_init(&async->ring) == 0) {
+ pthread_port_set_ring(&async->port, &async->ring);
+ pthread_attr_setdetachstate(&attrs, 1);
+ if ((pthread_create(&thread, &attrs, async_thread, NULL))
+ == 0)
+ res = TRUE;
} else
- syslog(LOG_NOTICE,
- "async_init_pth() - pth_msgport_create()");
- pth_attr_destroy(attrs);
- }
+ syslog(LOG_NOTICE, "async_init_pth() - pth_event()");
+ } else
+ syslog(LOG_NOTICE,
+ "async_init_pth() - pth_msgport_create()");
+ pthread_attr_destroy(&attrs);
}
if (!res) {
- if (async->ring) {
- pth_event_free(async->ring, PTH_FREE_ALL);
- async->ring = NULL;
- }
- if (async->port) {
- pth_msgport_destroy(async->port);
- async->port = NULL;
- }
+ if (pthread_port_valid(&async->port))
+ pthread_port_destroy(&async->port);
+ if (pthread_ring_valid(&async->ring))
+ pthread_ring_destroy(&async->ring);
pool_destroy(&async->apool);
}
} else
- DEBUG_PRINTF("async_init_pth", "Call async_init() first");
+ DEBUG_PRINTF("async_init_pthread", "Call async_init() first");
return (res);
}
/* Used by a client thread to initialize a context suitable to call the
* async_call() function with. Each thread needs one to use the device.
- * async_init() and async_init_pth() should have been called first.
+ * async_init() and async_init_pthread() should have been called first.
*/
static struct async_clenv *
async_open_clenv(void)
if (DEBUG_TRUE(async != NULL && POOL_VALID(&async->apool))) {
/* Optimized for speed using a pool of pre-allocated nodes */
- pth_mutex_acquire(&async->apool_lock, FALSE, NULL);
+ pthread_mutex_lock(&async->apool_lock);
aclenv = (struct async_clenv *)pool_alloc(&async->apool, FALSE);
- pth_mutex_release(&async->apool_lock);
+ pthread_mutex_unlock(&async->apool_lock);
} else
DEBUG_PRINTF("async_open_clenv",
"Call async_init() and async_init_pth() first");
static struct async_clenv *
async_close_clenv(struct async_clenv *aclenv)
{
- pth_mutex_acquire(&async->apool_lock, FALSE, NULL);
+ pthread_mutex_lock(&async->apool_lock);
pool_free((pnode_t *)aclenv);
- pth_mutex_release(&async->apool_lock);
+ pthread_mutex_unlock(&async->apool_lock);
return (NULL);
}
{
struct async_clenv *aclenv = (struct async_clenv *)pnode;
- if ((aclenv->port = pth_msgport_create("XXX(NULL)")) != NULL) {
- if ((aclenv->ring = pth_event(PTH_EVENT_MSG, aclenv->port))) {
- register char *ptr = (char *)aclenv;
+ if (pthread_port_init(&aclenv->port) == 0) {
+ if (pthread_ring_init(&aclenv->ring) == 0) {
+ register char *ptr = (char *)aclenv;
ptr += (size_t)OALIGN_CEIL(sizeof(struct async_clenv), long);
aclenv->msg = (struct async_msg *)ptr;
- aclenv->msg->msg.m_replyport = aclenv->port;
- aclenv->msg->msg.m_size = async->msg_len;
+ pthread_msg_init(&aclenv->msg->msg, &aclenv->port);
return TRUE;
}
- pth_msgport_destroy(aclenv->port);
+ pthread_port_destroy(&aclenv->port);
}
return FALSE;
{
struct async_clenv *aclenv = (struct async_clenv *)pnode;
- pth_event_free(aclenv->ring, PTH_FREE_ALL);
- pth_msgport_destroy(aclenv->port);
+ pthread_msg_destroy(&aclenv->msg->msg);
+ pthread_port_destroy(&aclenv->port);
+ pthread_ring_destroy(&aclenv->ring);
}
/* Send request */
aclenv->msg->func_id = function;
aclenv->msg->aclenv = aclenv;
- pth_msgport_put(async->port, &(aclenv->msg->msg));
+ pthread_msg_put(&async->port, &(aclenv->msg->msg));
/* Sleep until we get reply back, in a thread-safe manner.
* Note: This will permanently block the current thread if the
* we get a reply back seems a good choice here; The async functions
* should ensure to eventually return.
*/
- for (;;) {
- if ((pth_wait(aclenv->ring))) {
- if ((pth_event_occurred(aclenv->ring))) {
- /* We know that message pointer will be the same */
- if ((pth_msgport_get(aclenv->port)))
- break;
- }
- }
- }
+ while (pthread_msg_get(&aclenv->port) == NULL)
+ pthread_ring_wait(&aclenv->ring, NULL);
}
data.cnt = 0;
for (;;) {
/* Sleep until it is known that at least one node expired */
- pth_sleep((unsigned int)data.soonest);
+ pthread_sleep((unsigned int)data.soonest);
/* Tell our iterator function the current time and the maximum
* allowed time to wait to
*/
/* Lock ctable, reset expired nodes, garbage collect, and set
* data.soonest to the time of the soonest next expireing node.
*/
- pth_mutex_acquire(&ctable_lock, FALSE, NULL);
+ pthread_mutex_lock(&ctable_lock);
if (HASHTABLE_NODES(&ctable) > 0)
hashtable_iterate(&ctable, clnode_expire_thread_iterator, &data);
- pth_mutex_release(&ctable_lock);
+ pthread_mutex_unlock(&ctable_lock);
}
/* NOTREACHED */
- pth_exit(NULL);
+ pthread_exit(NULL);
return NULL;
}
if (rem != 0 && data->soonest > rem)
data->soonest = rem;
- /* If the cache is big, prevent from interfering with other threads */
- if ((data->cnt++) == 64) {
- data->cnt = 0;
- pth_yield(NULL);
- }
-
return TRUE;
}
Repositories / mmondor.git / commitdiff