+++ /dev/null
-/*
- openisis - an open implementation of the CDS/ISIS database
- Version 0.8.x (patchlevel see file Version)
- Copyright (C) 2001-2003 by Erik Grziwotz, erik@openisis.org
-
- This library is free software; you can redistribute it and/or
- modify it under the terms of the GNU Lesser General Public
- License as published by the Free Software Foundation; either
- version 2.1 of the License, or (at your option) any later version.
-
- This library is distributed in the hope that it will be useful,
- but WITHOUT ANY WARRANTY; without even the implied warranty of
- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- Lesser General Public License for more details.
-
- You should have received a copy of the GNU Lesser General Public
- License along with this library; if not, write to the Free Software
- Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-
- see README for more information
-EOH */
-
-/*
- $Id: lsv.c,v 1.14 2003/06/11 14:53:26 kripke Exp $
- OpenIsis server
-*/
-
-#include <sys/select.h>
-#include <sys/time.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <unistd.h>
-#include <fcntl.h>
-#include <netinet/in.h>
-#include <string.h> /* memcpy et al */
-#include <signal.h>
-#include <errno.h>
-
-
-#include "lsv.h"
-
-
-
-static Con *con[FD_SETSIZE]; /* usually 1024, but up to 1<<16 */
-#ifdef LSVTHR
-static pthread_mutex_t lsv_mutex_init = PTHREAD_MUTEX_INITIALIZER;
-static pthread_cond_t lsv_cond_init = PTHREAD_COND_INITIALIZER;
-static pthread_key_t lsv_key_wrk;
-#else
-static Wrk *lsv_wrk_single;
-#endif
-static int lsv_term;
-static siginfo_t lsv_sig;
-
-static const char * pdg[] = {
- "---",
- "NEW",
- "RED",
- "EOF"
-};
-static const char * stg[] = {
- "NEW",
- "CON",
- "INP",
- "SES",
- "WRK",
- "RUN",
- "COM",
- "DON",
- "RET"
-};
-
-static void lsv_sighand ( int sig, siginfo_t *info, void *ucontext )
-{
- if ( SIGINT == sig || SIGTERM == sig )
- lsv_term = !0;
- if ( info )
- lsv_sig = *info;
- else { /* ??? */
- lsv_sig.si_signo = sig;
- lsv_sig.si_pid = 0;
- }
- (void)ucontext;
-} /* lsv_sighand */
-
-
-static int lsv_enq ( Que *q, Con *c )
-{
- c->nxt = 0;
- q->len++;
- if ( q->tail ) {
- q->tail->nxt = c;
- q->tail = c;
- return 1;
- }
- q->head = q->tail = c;
- return 0;
-} /* lsv_enq */
-
-
-static void lsv_penq ( Srv *srv, Con *c )
-{
- Pool *work =
-#ifdef LSVTHR
- c->grp ? &srv->pool :
-#endif
- &srv->main;
- srv->jobs++;
- c->stg = LSV_STGWRK;
- if ( c->ses )
- c->ses->cur = c;
- lsv_enq( &work->que, c );
- sMsg( LOG_DEBUG, "enq req %d from %s qlen %d",
- LIO_FD&c->ios.file, c->nam, work->que.len );
-#ifdef LSVTHR
- if ( work->wait ) {
- Wrk *w = work->wait;
- work->wait = w->nxt;
- w->nxt = 0;
- srv->plen--;
- pthread_cond_signal( &w->todo );
- sMsg( LOG_DEBUG, "woke %d of %d", w->id, srv->plen );
- }
-#endif
-} /* lsv_penq */
-
-
-static Con *lsv_deq ( Que *q )
-{
- Con *c = q->head;
- if ( c ) {
- q->len--;
- q->head = c->nxt;
- c->nxt = 0;
- if ( q->tail == c )
- q->tail = 0;
- }
- return c;
-} /* lsv_deq */
-
-
-Wrk *svCur ()
-{
-#ifdef LSVTHR
- return (Wrk*)pthread_getspecific( lsv_key_wrk );
-#else
- return lsv_wrk_single;
-#endif
-}
-
-
-/**
- in threaded mode, run forever, waiting for jobs.
- non-threaded, do the jobs that are available and return
-*/
-static void *lsv_wrk ( void *arg )
-{
- Wrk *self = (Wrk*)arg;
- Srv *srv = self->srv;
- Con *job = 0;
- Pool *work =
-#ifdef LSVTHR
- self->id ? &srv->pool :
-#endif
- &srv->main;
-#ifdef LSVTHR
- sigset_t blk;
-
- /* block signals, they should go to the receiver */
- sigemptyset( &blk );
- sigaddset( &blk, SIGHUP );
- sigaddset( &blk, SIGINT );
- sigaddset( &blk, SIGTERM );
- sigaddset( &blk, SIGPIPE );
- pthread_sigmask( SIG_BLOCK, &blk, 0 );
-
- pthread_setspecific( lsv_key_wrk, arg );
- if ( self->id )
- srv->app( 0, LSV_APPINI );
-#else
- lsv_wrk_single = self;
- if ( !self->ses )
-#endif
- self->ses = sGet(); /* allocate session for thread */
-
- for (;/*ever*/;) {
- int ret;
-
- /* ****************************************
- critical section enter
- */
-#ifdef LSVTHR
- pthread_mutex_lock( &srv->mut );
-#endif
- if ( job ) { /* from last turn */
- sMsg( LOG_DEBUG, "thr %d enq job %s pdg %s",
- self->id, job->nam, pdg[job->pdg] );
- job->stg = LSV_STGRET;
- lsv_enq( &srv->recv, job );
-#ifdef LSVTHR
- if ( job->pdg ) /* wakeup receiver */
- write( srv->pip[1], &job->pdg, 1 );
-#endif
- if ( job->ses->que ) { /* immediatly reschedule queued job */
- Con *q = job->ses->que;
- job->ses->que = q->nxt;
- sMsg( LOG_VERBOSE, "dequeing con %s for ses '%s'",
- q->nam, job->ses->name );
- lsv_penq( srv, q );
- }
- }
- self->cur = 0;
- while ( !(job = lsv_deq( &work->que ))
- && !(LSV_SHUTDN & srv->flg)
- ) {
-#ifndef LSVTHR
- return self;
-#else
- self->nxt = work->wait;
- work->wait = self; /* pls pls lemme wrk */
- srv->plen++;
- self->waits++;
- sMsg( LOG_DEBUG, "thr %d waiting", self->id );
- /* wait releases the mutex */
- pthread_cond_wait( &self->todo, &srv->mut );
-#endif
- }
- self->cur = job;
- if ( job )
- job->stg = LSV_STGRUN;
-#ifdef LSVTHR
- pthread_mutex_unlock( &srv->mut );
-#endif
- /* ****************************************
- critical section leave
- */
-
- /* end of monday morning meeting blah blah -- start working */
- if ( ! job || (LSV_SHUTDN & srv->flg) )
- break;
- self->jobs++;
- if ( job->ses )
- sSet( job->ses );
- else
- job->ses = self->ses;
- job->ses->io[1] = &job->ios;
- if ( job->ses->res )
- CLRREC( job->ses->res );
- sMsg( LOG_DEBUG, "thr %d run job %s ses '%s'.%d",
- self->id, job->nam, job->ses->name, job->ses->accnt );
- ret = srv->app( job, LSV_APPRUN );
- sMsg( LOG_DEBUG, "thr %d got %d buf %d",
- self->id, ret, LIO_SAVAIL( &job->ios ) );
-
- /*
- lock only in order to create SMP memory barrier.
- on a single processor system, the byte should be safely readable
- by receiver without this lock.
- TOD: check alternatives:
- since receiver is checking the stage of active connections
- only in rare cases, it might be more efficient to guard either
- this set or the processing up to here with a per thread mutex.
- */
-#ifdef LSVTHR
- pthread_mutex_lock( &srv->mut );
-#endif
- job->stg = LSV_STGDON;
- job->ses->cur = 0;
-#ifdef LSVTHR
- pthread_mutex_unlock( &srv->mut );
-#endif
-
- ret = srv->prt( &job->ios, LIO_SCLOSE );
- job->ses->io[1] = 0;
- sMsg( LOG_DEBUG, "thr %d done job %s", self->id, job->nam );
- if ( job->ses != self->ses )
- sSet( self->ses ); /* reset session */
- }
-#ifndef LSVTHR
- if ( self->id )
- srv->app( 0, LSV_APPFIN );
-#endif
- return self;
-} /* lsv_wrk */
-
-
-static void *lsv_rcv ( void *arg )
-{
- Srv *srv = (Srv*)arg;
- fd_set fd;
- int fdlen = 0;
- int pending = 0;
- int ret = 0, sel, i;
-#ifdef LSVTHR
- int inlock = 0;
-#endif
-
- FD_ZERO( &fd );
- FD_SET( srv->lsn, &fd );
- fdlen = srv->lsn+1;
-#ifdef LSVTHR
- FD_SET( srv->pip[0], &fd );
- if ( fdlen < srv->pip[0]+1 )
- fdlen = srv->pip[0]+1;
-#endif
-
- for (;/*ever*/;) {
- struct sockaddr_in npeer; /* of new connection */
- int nsock = -1; /* new socket */
- Con *c;
- fd_set ready = fd;
- Tm expire;
-
- /* select on read only; writing is done in workers and
- exceptions (i.e. telnet OOB data) are not used
- */
-#ifndef LSVTHR
- if ( srv->recv.len ) { /* instead of listening on pip */
- memset( &ready, 0, sizeof(ready) );
- sel = 0;
- } else
-#endif
- if ( 0 > (sel = select( fdlen, &ready, 0, 0, 0 )) ) {
- if ( EINTR == errno ) {
- sMsg( LOG_DEBUG, "select woken by signal" );
- } else {
- ret = sMsg( LOG_SYSERR, "select" );
- goto done;
- }
- }
- if ( lsv_sig.si_signo ) {
- sMsg( LOG_DEBUG, "SIG%s (%d) from pid %d",
- SIGHUP==lsv_sig.si_signo ? "HUP" :
- SIGINT==lsv_sig.si_signo ? "INT" :
- SIGTERM==lsv_sig.si_signo ? "TERM" :
- SIGPIPE==lsv_sig.si_signo ? "PIPE" : "?",
- lsv_sig.si_signo, lsv_sig.si_pid );
- if ( 1 /*SIGHUP==lsv_sig.si_signo*/ ) {
- for ( i=FD_SETSIZE; i--; )
- if ( con[i] ) {
- c = con[i];
- sMsg( LOG_INFO,
- "con %3d peer %s stg %s prt %d app %d ios %d %s (0x%x)",
- i, c->nam, stg[c->stg], c->prt, c->app,
- c->ios.pos + c->ios.b.done,
- LIO_SISOPEN(&c->ios) ? "opn" : "clo", c->ios.file );
- }
- }
- if ( lsv_term ) /* even if it wasn't the last signal delivered */
- goto done;
- lsv_sig.si_signo = 0;
- /* must NOT access the set after interrupt,
- probably all fds are set
- */
- continue;
- }
- timeUpd( &srv->tim );
- timeGtfm( srv->gtm, &srv->tim );
- expire.millis = srv->tim.millis - srv->sto*LLL(1000);
-
-#ifdef LSVTHR
- pthread_mutex_lock( &srv->mut ); /* back in locked land */
- inlock = !0;
-#endif
-
- /* -- we're the acceptor */
- if ( FD_ISSET( srv->lsn, &ready ) ) { /* new connection */
- unsigned /*socklen_t is broken*/ alen = sizeof(npeer);
- nsock = accept( srv->lsn, (struct sockaddr*)&npeer, &alen );
- if ( 0 > nsock ) {
- ret = sMsg( LOG_SYSERR, "accept" );
- goto done;
- }
- if ( FD_SETSIZE <= nsock ) {
- ret = sMsg( ERR_BADF,
- "socket %d >= FD_SETSIZE %d", nsock, FD_SETSIZE );
- goto done;
- }
- if ( sizeof(npeer) != alen ) {
- ret = sMsg( ERR_INVAL, "bad peer len %d", alen );
- goto done;
- }
- /* ok */
- /* setsockopt
- SO_SNDTIMEO fixed value on linux
- SO_OOBINLINE ... hmm, don't care
- SO_LINGER off
- */
- /* prepare con */
- if ( !(c = con[nsock]) ) {
- c = con[nsock] = mAlloc( sizeof(*c) );
- c->ios.file = nsock; /* preliminary -- further setup below */
- }
- c->pdg = LSV_PDGNEW;
- switch ( c->stg ) {
- case LSV_STGCON: /* still in control of receiver */
- case LSV_STGINP:
- sMsg( ERR_IDIOT, "bad recv stage %d on new con", c->stg );
- c->stg = LSV_STGNEW;
- case LSV_STGNEW: /* clean idle socket -- pass to receiver */
- lsv_enq( &srv->recv, c );
- break;
- case LSV_STGSES: /* still in control of receiver */
- case LSV_STGWRK:
- sMsg( ERR_IDIOT, "bad worker stage %d on new con", c->stg );
- case LSV_STGRUN: /* worker will queue it */
- /* hmmm ... should have COM ... */
- case LSV_STGCOM:
- case LSV_STGDON:
- case LSV_STGRET: /* worker has queued it */
- pending++;
- break;
- }
- srv->conn++;
- FD_CLR( srv->lsn, &ready ); /* skip in fd check */
- } /* new connection */
-
-#ifdef LSVTHR
- if ( FD_ISSET( srv->pip[0], &ready ) ) { /* pending events */
- char trash[LSV_NUMWRK];
- read( srv->pip[0], trash, sizeof(trash) );
- FD_CLR( srv->pip[0], &ready );
- sel--;
- }
-#endif
-
- /* switching roles -- now we're the receiver */
- while ( (c = lsv_deq( &srv->recv )) ) { /* care for ready connections */
- int sock = LIO_FD & c->ios.file;
- sMsg( LOG_DEBUG, "deq %d from %s pdg %s",
- sock, c->nam, pdg[c->pdg] );
- if ( c->pdg && c->stg )
- pending--;
- if ( !(LIO_IN & c->ios.file) && (LIO_RDWR & c->ios.file) ) {
- /* cleanup closed socks */
- if ( c != con[sock] ) {
- ret = sMsg( ERR_IDIOT,
- "deq bad sock %d (0x%x)", sock, c->ios.file );
- goto done; /* PANIC time */
- }
- sMsg( LOG_DEBUG, "clos%s sock %d",
- (LIO_OUT & c->ios.file) ? "ing" : "ed", sock );
- if ( (LIO_OUT & c->ios.file) )
- lio_close( &c->ios.file, LIO_INOUT );
- c->ios.file &= ~LIO_RDWR; /* clear also, so we're not closing again */
- FD_CLR( sock, &fd );
- FD_CLR( sock, &ready ); /* probably an EOF to read */
- if ( fdlen == sock+1 )
- while ( fdlen
- /* && !FD_ISSET( fdlen-1, &fd ) risk of muting tmp. disabled */
- && (!con[fdlen-1] || !LIO_SISOPEN(&con[fdlen-1]->ios))
- )
- fdlen--;
- }
- if ( LSV_PDGNEW == c->pdg ) { /* new connection */
- struct sockaddr_in peer; /* of new connection */
- if ( nsock == sock ) /* the one accepted just above */
- peer = npeer;
- else {
- unsigned alen = sizeof(peer);
- if ( getpeername( sock, (struct sockaddr*)&peer, &alen )
- || sizeof(peer) != alen
- ) {
- ret = sMsg( LOG_SYSERR, "getpeername alen %d", alen );
- goto done;
- }
- }
- c->srv = srv;
- c->ses = 0;
- {
- unsigned host = (unsigned)ntohl( peer.sin_addr.s_addr );
- unsigned short port = (unsigned short)ntohs( peer.sin_port );
- char *p = c->nam;
- c->host = host;
- memset( c->nam, 0, sizeof(c->nam) );
- p += u2a( p, 0xff & (host >> 24) ); *p++ = '.';
- p += u2a( p, 0xff & (host >> 16) ); *p++ = '.';
- p += u2a( p, 0xff & (host >> 8) ); *p++ = '.';
- p += u2a( p, 0xff & (host ) ); *p++ = ':';
- p += u2a( p, port );
- /* *p++ = '@'; p += u2a( p, sock ); */
- }
- c->con++; /* never mind overflow ... */
- sMsg( LOG_INFO,
- "connect %d on %d from %s", c->con+1, sock, c->nam );
- LIO_SINIT( &c->ios, srv->prt, c->nam,
- sock | LIO_INOUT|LIO_RDWR|LIO_SOCK );
- LIO_BINIT( &c->flt );
- FD_SET( sock, &fd );
- if ( fdlen < sock+1 )
- fdlen = sock+1;
- c->bin = 0; /* do not reset on every request */
- }
- if ( ! (LIO_IN & c->ios.file) )
- c->stg = LSV_STGNEW;
- else {
- c->stg = LSV_STGCON;
- if ( ! FD_ISSET( sock, &fd ) ) {
- sMsg( LOG_DEBUG, "reenabling %s", c->nam );
- FD_SET( sock, &fd );
- if ( fdlen < sock+1 )
- fdlen = sock+1;
- }
- }
- c->grp = c->pdg = c->prt = c->app = 0;
- /* prepare for new request */
- c->ios.pos = 0;
- c->ios.b.done = c->ios.b.fill = 0;
- if ( c->req )
- CLRREC( c->req );
- else {
- /* this is done in the receiver rather than the acceptor
- since in multi-receiver model, receiver owns the req
- */
- c->req = mAlloc( LSV_BUFSIZ );
- OPENISIS_INITREC( c->req, LSV_BUFSIZ, 64 );
- }
- if ( !(LSV_CONSES & srv->flg) )
- c->ses = 0;
- } /* prepare connections */
-
-
- for ( i=fdlen; sel && i--; ) { /* read from the ready */
- Field *sid;
- int got;
- if ( ! FD_ISSET( i, &ready ) )
- continue;
- sel--;
- c = con[i];
- if ( ! c ) {
- ret = sMsg( ERR_IDIOT, "ran on empty con %d", i );
- goto done;
- }
- if ( LSV_STGINP < c->stg ) {
- sMsg( LOG_DEBUG, "con %d busy stage %d!", i, c->stg );
- c->pdg = LSV_PDGRED;
- pending++;
- FD_CLR( i, &fd ); /* mute this */
- continue;
- }
- c->ios.b.done = 0;
- got = lio_read( &c->ios.file, c->ios.b.c, sizeof(c->ios.b.c) );
- if ( 0 >= got ) {
- if ( -ERR_EOF != got )
- goto err;
- sMsg( LOG_DEBUG, "EOF on 0x%x line %d from %s",
- c->ios.file, c->req->len, c->nam );
- FD_CLR( i, &fd );
- if ( ! c->req->len ) /* initial eof */
- goto eof; /* close immediatly */
- /* lio_close( &c->ios.file, LIO_IN ); */
- c->ios.file &= ~LIO_IN; /* don't shut */
- c->pdg = LSV_PDGEOF; /* so it will be dequed and closed */
- pending++;
- got = 0; /* tell proto to close ... */
- }
- if ( ! c->req->len ) { /* stage should be CON */
- RADDS( c->req, -LSV_PEER, c->nam, 1 );
- sMsg( LOG_DEBUG, "req on %d from %s",
- LIO_FD&c->ios.file, c->nam );
- }
- c->stg = LSV_STGINP;
- c->ios.b.fill = got;
- got = srv->prt( &c->ios, LIO_SPUSH );
- sMsg( LOG_DEBUG, "got 0x%x from %s", got, c->nam );
- if ( ! got && (LIO_IN & c->ios.file) ) /* ok: more next time */
- continue;
- if ( 0 > got ) /* nok */
- goto err;
- got = c->srv->app( c, LSV_APPGOT );
- if ( 0 > got ) /* nok */
- goto err;
- if ( !(LSV_CONSES & srv->flg)
- && c->req && (sid = rGet( c->req, -LSV_SID, 0 ))
- ) {
- c->ses = cSesByName( (char*)sid->val, sid->len, &srv->tim, &expire );
- if ( ! c->ses ) { /* sad sad sad */
- sMsg( LOG_ERROR, "out of sessions" );
- goto err; /* drop connection */
- }
- if ( ! c->ses->accnt
- && (! c->ses->prop || ! c->ses->prop->len) /* overflow paranoia */
- )
- got = c->srv->app( c, LSV_APPSES );
- if ( 0 > got ) /* nok */
- goto err;
- }
- c->tim = srv->tim;
- RADDS( c->req, -LSV_TIME, srv->gtm, 1 );
- if ( c->ses ) {
- sMsg( LOG_DEBUG, "req %s for ses '%s' %s",
- c->nam, c->ses->name, c->ses->cur ? "busy" : "idle" );
- c->ses->accnt++;
- if ( c->ses->cur || c->ses->que ) { /* oh - busy !? */
- Con **cpp = &c->ses->que;
- int l = 0;
- for ( ; *cpp; cpp = &(*cpp)->nxt )
- if ( 4 < ++l ) {
- sMsg( LOG_WARN,
- "too many con queing on ses '%s' - dropping %s",
- c->ses->name, c->nam );
- goto err;
- }
- sMsg( LOG_VERBOSE, "queing con %s for ses '%s'",
- c->nam, c->ses->name );
- c->nxt = 0;
- *cpp = c;
- c->stg = LSV_STGSES;
- continue;
- }
- }
- /* yo - got it */
- lsv_penq( srv, c );
- continue;
- err:
- sMsg( ERR_BADF, "err 0x%x from %s", got, c->nam );
- eof:
- lio_close( &c->ios.file, LIO_INOUT );
- c->stg = LSV_STGNEW;
- FD_CLR( i, &fd );
- } /* read from the ready */
-
- srv->turn++;
- srv->busy += srv->nwr - srv->plen;
-#ifdef LSVTHR
- srv->wlen += srv->pool.que.len;
-
- sMsg( LOG_INFO,
- "\n==== %s\nturn %d: %d jobs for %d idle workers %d pending",
- srv->gtm, srv->turn, srv->pool.que.len, srv->plen, pending );
- for ( i=srv->nwr; i--; )
- if ( srv->wrk[i].cur )
- sMsg( LOG_INFO, "thr %d serving %d from %s",
- i, LIO_FD&srv->wrk[i].cur->ios.file, srv->wrk[i].cur->nam );
-
- inlock = 0;
- pthread_mutex_unlock( &srv->mut );
-#else
- lsv_wrk( srv->wrk );
-#endif
- } /* for(ever) */
-done:
-#ifdef LSVTHR
- if ( inlock )
- pthread_mutex_unlock( &srv->mut );
- srv->flg |= LSV_SHUTDN;
- for ( i=srv->nwr; i--; )
- pthread_cond_signal( &srv->wrk[i].todo );
-#endif
- return srv;
-} /* lsv_rcv */
-
-
-int svRun ( Srv *srv, const char *addr )
-{
- static const int yes = !0;
- struct sigaction siga;
- struct sockaddr_in sa; /* server address */
- int ret = 0, i;
-
- /* assume we're the main-thread
- -- try to make sure we get the controlling session
- */
- sGet();
-
- /* set defaults */
- if ( ! srv->sto )
- srv->sto = 30*60; /* 30 minutes session timeout */
-#ifndef LSVTHR
- srv->nwr = 1;
-#else
- if ( ! srv->nwr )
- srv->nwr = LSV_NUMWRK/2; /* half the max */
- else if ( LSV_NUMWRK < srv->nwr )
- srv->nwr = LSV_NUMWRK; /* max */
-
- srv->mut = lsv_mutex_init;
-#endif
- /* prepare signals */
- memset( &siga, 0, sizeof(siga) );
- siga.sa_sigaction = lsv_sighand;
- sigemptyset( &siga.sa_mask );
- sigaddset( &siga.sa_mask, SIGHUP );
- sigaddset( &siga.sa_mask, SIGINT );
- sigaddset( &siga.sa_mask, SIGTERM );
- sigaddset( &siga.sa_mask, SIGPIPE );
- siga.sa_flags = SA_SIGINFO;
- sigaction( SIGHUP, &siga, 0 );
- sigaction( SIGINT, &siga, 0 );
- sigaction( SIGTERM, &siga, 0 );
- sigaction( SIGPIPE, &siga, 0 );
-
- /* prepare socket */
- srv->lsn = socket( PF_INET, SOCK_STREAM, IPPROTO_TCP/*6*/ );
- /* fcntl( srv->lsn, O_NONBLOCK ); not needed */
- sa.sin_family = AF_INET;
- sa.sin_port = htons( addr ? a2i( addr, -1 ) : 8080 );
- sa.sin_addr.s_addr =
-#ifdef LSV_LOCALHOST_ONLY
- htonl(INADDR_LOOPBACK);
-#else
- INADDR_ANY;
-#endif
- if ( setsockopt( srv->lsn, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes) ) ) {
- ret = sMsg( LOG_SYSERR, "setsockopt" );
- goto inierr;
- }
- if ( bind( srv->lsn, (struct sockaddr*)&sa, sizeof(sa) ) ) {
- ret = sMsg( LOG_SYSERR, "bind on port %u", htons(sa.sin_port) );
- goto inierr;
- }
- if ( listen( srv->lsn, 64/*backlog*/ ) ) {
- ret = sMsg( LOG_SYSERR, "listen" );
- goto inierr;
- }
-#ifdef LSVTHR
- /* prepare pending event pipe */
- if ( pipe( srv->pip ) ) {
- ret = sMsg( LOG_SYSERR, "pipe" );
- goto inierr;
- }
-
- if ( pthread_key_create( &lsv_key_wrk, 0 ) ) {
- ret = sMsg( LOG_SYSERR, "key_create" );
- goto inierr;
- }
-#endif
- /* setup workers */
- memset( srv->wrk, 0, sizeof(srv->wrk) );
- for ( i=srv->nwr; i--; ) {
- srv->wrk[i].id = i;
- srv->wrk[i].srv = srv;
- srv->wrk[i].nxt = 0;
- srv->wrk[i].ses = cSession(0);
- memcpy( srv->wrk[i].ses->name, "wrk", 3 );
- u2a( srv->wrk[i].ses->name+3, i );
-#ifdef LSVTHR
- srv->wrk[i].todo = lsv_cond_init;
- if ( i )
- pthread_create( &srv->wrk[i].thr, 0, lsv_wrk, srv->wrk+i );
-#endif
- }
-#ifndef LSVTHR
- lsv_rcv( srv );
-#else
- /* start the server thread */
- pthread_create( &srv->rcv, 0, lsv_rcv, srv );
- /* run main worker */
- srv->wrk[0].thr = pthread_self();
- lsv_wrk( srv->wrk );
-
- /* cleanup */
- pthread_join( srv->rcv, 0 );
- for ( i=srv->nwr; i--; )
- pthread_join( srv->wrk[i].thr, 0 );
-#endif
- for ( i=srv->nwr; i--; )
- sMsg( LOG_INFO, "worker %d had %d jobs %d waits",
- i, srv->wrk[i].jobs, srv->wrk[i].waits );
- sMsg( LOG_INFO,
- "server had %d jobs on %d connections %d turns"
- " avg queue len %.02f workers %.02f",
- srv->jobs, srv->conn, srv->turn,
- srv->wlen/srv->turn, srv->busy/srv->turn );
-inierr:
- close( srv->lsn );
- return ret;
-} /* svRun */
-
-
-/** helper, probably should go to lio.
- cat l bytes from c to buf b flushing as needed.
- if ! c, empty buf b.
-*/
-int lio_sout ( Ios *s, Buf *b, const char *c, unsigned l )
-{
- if ( ! b )
- b = &s->b;
- do {
- if ( LIO_BUFSIZ/2 < b->fill && b->done ) { /* purge */
- if ( b->done < b->fill )
- memmove( b->c, b->c + b->done, b->fill - b->done );
- s->pos += b->done;
- b->fill -= b->done;
- b->done = 0;
- }
- if ( c && l && b->fill < LIO_BUFSIZ ) {
- unsigned cp = LIO_BUFSIZ - b->fill;
- if ( cp > l )
- cp = l;
- memcpy( b->c + b->fill, c, cp );
- b->fill += cp;
- c += cp;
- l -= cp;
- if ( !l )
- return 0;
- }
- if ( b->fill > b->done ) {
- int wr = lio_write( &s->file, b->c + b->done, b->fill - b->done );
- if ( 0 > wr )
- return wr;
- if ( ! wr )
- return -ERR_BUSY;
- b->done += wr;
- if ( b->done > b->fill ) {
- b->done = b->fill = 0;
- return -ERR_IDIOT;
- }
- }
- if ( b->done >= b->fill ) {
- s->pos += b->done;
- b->done = b->fill = 0;
- }
- } while ( c ? l : (unsigned)(b->fill > b->done) );
- return 0;
-} /* lio_sout */
-
-
-
-#define LF 10 /* LineFeed a.k.a. newline - '\n' isn't really well defined */
-#define TAB 9 /* horizontal, that is */
-#define VT 11 /* vertical, used as newline replacement */
-static const char lf = LF;
-static const char tab = TAB;
-static const char vt = VT;
-
-/* the plain protocol */
-int svPlain ( Ios *s, int op )
-{
- Field *f;
- Con *c = (Con*)s;
- switch ( op ) {
- case LIO_SPUSH: {
- int l = s->b.fill - s->b.done;
- unsigned char *b = s->b.c + s->b.done;
- unsigned char *end = b+l, *v, *p;
- if ( ! l ) { /* EOF: done */
- if ( ! c->prt ) /* ok */
- return 1;
- /* last field wasn't closed by LF */
- return sMsg( ERR_INVAL, "no EOL from %s", c->nam );
- }
- if ( c->prt )
- RSPACE( c->req, l, !0 );
- /* add text lines */
- while ( b<end ) {
- int conti = 0;
- switch ( c->prt ) {
- case 0: /* at beginning of line -- start new field */
- if ( LF == *b ) /* empty line */
- return 1;
- if ( TAB != *b || !c->req->len )
- RADD( c->req, 0,0,end-b, !0 );
- else { /* binary mode continuation line */
- conti = 1;
- if ( ! c->bin ) {
- sMsg( LOG_INFO,
- "detected binary mode on con %s", c->nam );
- c->bin = 1;
- }
- RSPACE( c->req, end-b, !0 );
- }
- if ( ! c->req )
- return -ERR_NOMEM;
- c->prt = 1;
- case 1: /* add to last field */
- f = c->req->field + c->req->len-1;
- v = (unsigned char*)f->val;
- p = v + f->len;
- if ( conti ) {
- *p++ = LF;
- b++;
- }
- if ( c->bin ) {
- for ( ; b<end && LF != (*p = *b++); p++ )
- ;
- } else
- for ( ; b<end && LF != (*p = *b++); p++ )
- if ( VT == *p ) /* convert VTABs */
- *p = LF; /* back to newlines */
- c->req->used += (p - v) - f->len;
- f->len = p - v;
- if ( LF == b[-1] ) {
- int ret = a2il( f->val, f->len, &f->tag );
- if ( ret ) {
- if ( ret < f->len && TAB == v[ret] )
- ret++;
- if ( ret < f->len )
- memmove( v, v+ret, f->len - ret );
- f->len -= ret;
- }
- ret = c->srv->app( c, LSV_APPARG );
- c->prt = 0;
- if ( ret )
- return ret;
- }
- sMsg( LOG_INFO, "prs from %s: [%2d] %3d = '%.*s'",
- c->nam, c->req->len-1, f->tag, f->len, f->val );
- }
- }
- return 0;
- } /* case LIO_SPUSH */
- case LIO_SFLUSH:
- if ( s->b.fill <= s->b.done )
- return 0;
- case LIO_SCLOSE:
- if ( !(LIO_OUT & s->file) )
- return -ERR_INVAL;
- /*
- this switch is not protected, since receiver doesn't care
- */
- if ( c->stg < LSV_STGCOM )
- c->stg = LSV_STGCOM;
- if ( ! s->pos+s->b.done && c->ses->res ) {
- char num[32];
- int n = c->ses->res->len;
- sMsg( LOG_INFO, "com %d fields to %s", c->ses->res->len, c->nam );
- for ( f = c->ses->res->field; n--; f++ ) {
- const char *v = f->val, *nl;
- int vl = f->len;
- int nlen = i2a( num, f->tag );
- num[nlen++] = TAB;
- lio_sout( s, &c->flt, num, nlen );
- while ( vl && (nl = memchr(v,'\n',vl)) ) {
- int ll = 1+(nl-v);
- if ( c->bin ) {
- lio_sout( s, &c->flt, v, ll );
- lio_sout( s, &c->flt, &tab, 1 );
- } else {
- lio_sout( s, &c->flt, v, ll-1 );
- lio_sout( s, &c->flt, &vt, 1 );
- }
- v += ll;
- vl -= ll;
- }
- if ( vl )
- lio_sout( s, &c->flt, v, vl );
- lio_sout( s, &c->flt, &lf, 1 );
- }
- }
- if ( s->b.fill )
- s->b.done = s->b.fill = 0;
- /*
- {
- lio_sout( s, &c->flt, 0, 0 );
- lio_stdio( s, LIO_SFLUSH );
- }
- */
- if ( LIO_SCLOSE == op )
- lio_sout( s, &c->flt, &lf, 1 );
- lio_sout( s, &c->flt, 0, 0 ); /* finally flush */
- return 0;
- }
- return lio_stream( s, op ); /* hmm -- SPURGE ? */
-} /* svPlain */
-
-
-/* the echo application */
-int svEcho ( Con *c, int task )
-{
-#ifdef LSVECHO_DELAY
- static Tm tm = { LLL(50) };
-#endif
- Field *f;
- int i;
- if ( task ) { /* preparing: we don't care */
- if ( LSV_APPGOT == task )
- c->grp = 1; /* don't require mainthead */
- return 0;
- }
- /* run it */
- sMsg( LOG_INFO, "echo %d: %d fields to %s",
- svCur()->id, c->req->len, c->nam );
-#ifdef LSVECHO_DELAY
- timeSleep( &tm );
-#endif
- for ( i=c->req->len, f = c->req->field; i--; f++ ) {
- RADDF( c->ses->res, f, !0 );
- /*
- plain protocol actually does not support plain output ...
- sMsg( c->ses, 0, "%d = '%.*s'\n", f->tag, f->len, f->val );
- */
- }
- return 0;
-} /* svEcho */
-
-#if 0
-int main ( int argc, const char **argv )
-{
- Srv echo;
-
- memset( &echo, 0, sizeof(echo) );
- echo.prt = lsv_plain;
- echo.app = lsv_echo;
- cOpen(0);
- cLog( LOG_DEBUG, 0 );
- return svRun( &echo, 1<argc ? argv[1] : 0 );
-}
-#endif
projects / webpac / blobdiff