iframe-proxy | Sunbelt Computer Software

693 lines (584 loc) · 20.5 KB
#ifdef INFORMATION
Copyright (C) 2011-2012 by Outfit7
Further modifed by Bruce Wilcox 2014-2022
Released under Bruce Wilcox License as follows:
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#ifndef DISCARDSERVER
#ifdef EVSERVER
Copyright (C) 2011-2012Outfit7
by Igor Lautar <igor.lautar@outfit7.com>
Server implementation uses libev to trigger requests. Each request on a core is single-threaded
so there are no concurrency issues.
To compile, define EVSERVER during compilation, otherwise old pthread implementation is used.
Server listener:
 - listener socket is created in evsrv_init()
 - listener socket is added to libev for read event
Accepting requests:
 - libev triggers evsrv_accept
 - handler tries to accept as many connections as possible, creating Client_t instance for each
 - each client socket is registered in libev for read event
Handling client:
 - when data is available on client socket, client_read is called
 - it reads available data, if it is not enough it goes back to ev for more
 - if request is complete, it prepares the buffers and performs the chat
 - it immediatelly tries to send chat response, if socket is busy (EAGAIN) it adds write listener to ev
 - if write listener is triggered, client data is written
 - if there is still some data to write, write listener is added to ev again, until all data is written
 - once data is written, client instance is deregistered from libev, socket closed and object destroyed
#include "common.h"
#ifdef WIN32
	#include "ev.h"
	#include "evserver/ev.h"
#include "evserver.h"
#include <vector>
#include <algorithm>
extern char serverLogfileName[1000];
extern char dbTimeLogfileName[1000];
extern bool serverctrlz;
extern int cs_qsize;
#define CLIENT_CHUNK_LENGTH 4*1024
#define HIDDEN_OVERLAP 103	// possible concealed data
#define HIDDEN_OFFSET 3 // past 2 ctrl z's
// server stuff
string interface_g;
int port_g;
int listen_queue_length_g = 16*1024; // 16 k
int qsize = 0;
int srv_socket_g = -1;
// EV stuff
struct ev_loop *l_g = 0;
ev_io ev_accept_r_g;
ev_timer tt_g;
static int ev_pending = 0;
static int ev_max = 0;
#ifdef EVSERVER_FORK
// child monitors
#define MAX_CHILDREN_D 50
ev_child children_g[MAX_CHILDREN_D];
int no_children_g = 0;
int cur_children_g = 0;
bool parent_g = true;
static void evsrv_accept(EV_P_ ev_io *w, int revents);
static void client_read(EV_P_ ev_io *w, int revents);
static void client_write(EV_P_ ev_io *w, int revents);
static void evsrv_child_died(EV_P_ ev_child *w, int revents);
struct Client_t;
int evsrv_do_chat(Client_t *client);
typedef vector<char> Buffer_t;
#define ITER_TO_OFFSET(ctr, i) (&ctr[0] + (i - ctr.begin()))
struct Client_t
    char magic[9];
    int fd;
    int q; // how many in q ahead of this
    ev_io ev_r;
    ev_io ev_w;
    struct ev_loop *l;
    bool requestValid;
    string ip;
    char* bot;
    char* message;
    char* user;
    char* data = NULL;
    Buffer_t incomming;
	uint64 starttime = ElapsedMilliseconds(); 
    uint64 startnltime = 0;
    uint64 endnltime = 0;
    uint64 qtime = 0;
    uint64 nltime = 0;
    void Qdown()
        --ev_pending;
        delete this;
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wreorder-ctor"
    Client_t(int fd, struct ev_loop *l_p, int queue) : fd(fd), l(l_p), q(queue),requestValid(false)
        strcpy(this->magic, "deadbeef");
        ev_io_init(&this->ev_r, client_read, this->fd, EV_READ);
        ev_io_init(&this->ev_w, client_write, this->fd, EV_WRITE);
        this->ev_r.data = this;
        this->ev_w.data = this;
        ev_io_start(this->l, &this->ev_r);
#pragma GCC diagnostic pop
    ~Client_t() // if child dies, this destructor is not called
		if (this->data) myfree(this->data); 
		this->data = NULL;
        if (ev_is_active(&this->ev_r))  ev_io_stop(this->l, &this->ev_r);
        if (ev_is_active(&this->ev_w))  ev_io_stop(this->l, &this->ev_w);
        close(this->fd);
    void prepare_for_next_request()
        this->incomming.clear();
        this->requestValid = false;
        this->bot = 0;
        this->message = 0;
        this->user = 0;
        if (this->data) *this->data = 0;
        if (!ev_is_active(&this->ev_r))  ev_io_start(this->l, &this->ev_r);
        if (ev_is_active(&this->ev_w))   ev_io_stop(this->l, &this->ev_w);
    string get_foreign_address()
        sockaddr_in addr;
        unsigned int addr_len = sizeof(addr);
        if (getpeername(this->fd, (sockaddr *) &addr,(socklen_t *) &addr_len) < 0)  return "";
        return inet_ntoa(addr.sin_addr);
    int recv_data()
        if (this->requestValid)
            ReportBug((char*)"recv called although we got whole request, should process it first\r\n");
            return -1;
        int read_size = CLIENT_CHUNK_LENGTH;
        size_t dataEnd = this->incomming.size();
        this->incomming.resize(dataEnd + read_size);
        char *read_ptr = &this->incomming[dataEnd];
        int r = recv(this->fd, read_ptr, read_size, 0);
        // make buffer as long as actual data
        if (r > 0)  this->incomming.resize(dataEnd + r);
        return r;
    int send_data()
		size_t len = strlen(this->data);
		if (serverctrlz) len += 3; // send end marker for positive confirmation of completion
        // try to send, if we get eagain error, schedule it for later
        int r = send(this->fd, this->data, len, 0);
        if (r < 0) 
            if (errno == EAGAIN)
                ev_io_start(this->l, &this->ev_w);
                return 0;
            ReportBug("evserver: send_data(%d) could not send, errno: %s\r\n", len,strerror(errno));
			(*printer)( "evserver: send_data(%d) could not send, errno: %s", len,strerror(errno));
			return -1;
        if (r < (int)len) 
            ev_io_start(this->l, &this->ev_w);
			memmove(data,data+r,(len-r) + HIDDEN_OVERLAP);
            return 0;
        // sent all data
        uint64 sendtime = ElapsedMilliseconds() - this->endnltime; // duration of send
        if (sendtime >= 1000)
            ReportBug("INFO: Excess Send Time %d %s\r\n", (int)(sendtime), message);
        this->prepare_for_next_request();
        return 1;
    void prepare_for_chat()
        if (this->data) *this->data = 0;
        this->ip = this->get_foreign_address();
        if (this->ip.length() == 0) 
            Log(SERVERLOG, "evserver: prepare_for_chat() could not get ip for client: %d\r\n", this->fd);
			(*printer)("evserver: prepare_for_chat() could not get ip for client: %d\r\n", this->fd);
    int received_request() {
        int nulls = count(this->incomming.begin(), this->incomming.end(), 0);
		if ((nulls) < 3)  return 0;
        if ((nulls) > 3)   return -1;
        this->requestValid = true;
        user = &this->incomming[0];
        if (strlen(user) == 0)  return -1;
		Buffer_t::iterator next_null;
		next_null = find(this->incomming.begin(), this->incomming.end(), 0);
		bot = ITER_TO_OFFSET(this->incomming, next_null + 1);
		next_null = find(next_null + 1, this->incomming.end(), 0);
		message = ITER_TO_OFFSET(this->incomming, next_null + 1);
        // since we received complete request, we will stop reading from client socket until we process it
        ev_io_stop(this->l, &this->ev_r);
        return 1;
static int settcpnodelay(int fd)
    int on = 1;
#ifdef WIN32
    return setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char*) &on, sizeof(on));
    return setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (void*) &on, sizeof(on));
static int setnonblocking(int fd)
    // non blocking (fix for windows)
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags == -1) {
        Log(SERVERLOG, "evserver: setnonblocking() fcntl(F_GETFL) failed, errno: %s\r\n", strerror(errno));
		(*printer)( "evserver: setnonblocking() fcntl(F_GETFL) failed, errno: %s\r\n", strerror(errno));
		return -1;
    if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) 
        Log(SERVERLOG, "evserver: setnonblocking() fcntl(F_SETFL) failed, errno: %s\r\n", strerror(errno));
		(*printer)( "evserver: setnonblocking() fcntl(F_SETFL) failed, errno: %s\r\n", strerror(errno));
		return -1;
    return 1;
#ifdef EVSERVER_FORK
int fork_child(ev_child *child_watcher = 0)
    string curDir;
    pid_t pid = 0;
    pid = fork();
    if (pid < 0) {
		Log(SERVERLOG, "evserver: fork failed, errno %d\r\n", errno);
		(*printer)( "evserver: fork failed, errno %d\r\n", errno);
		return -1;
    if (pid > 0) {
        // parent
        if (!child_watcher) {
            child_watcher = &children_g[cur_children_g];
            cur_children_g++;
		else   ev_child_stop(l_g, child_watcher);
        ev_child_init(child_watcher, evsrv_child_died, pid, 0);
        ev_child_start(l_g, child_watcher);
        currentBeforeLayer = LAYER_BOOT;
        return 0;
    int b = false;
    if (!b)   sleep(1);
    // child
    ev_loop_fork(l_g);
#ifndef DISCARDJSONOPEN
    CurlShutdown();
    for (int i = 0; i < cur_children_g; i++)   ev_child_stop(l_g, &children_g[i]);
    cur_children_g = 0;
    parent_g = false;
    return 1;
char* currentDateTime() {
    struct tm ptm;
    return GetTimeInfo(&ptm);
static void evsrv_child_died(EV_P_ ev_child *w, int revents) {
    char* timestamp = currentDateTime();
    Log(SERVERLOG, "%s - evserver: evsrv_child_died [pid: %d]\r\n", timestamp,w->pid);
	(*printer)("%s - evserver: evsrv_child_died [pid: %d]\r\n", timestamp, w->pid);
	int r = fork_child(w);
	if (r < 0)
		ReportBug( "  %s - evserver: could not re-spawn child after it died [pid: %d]\r\n", timestamp, w->pid);
		(*printer)("  %s - evserver: could not re-spawn child after it died [pid: %d]\r\n", timestamp,w->pid);
	else if (r == 1)
		// socket listener
		ev_io_init(&ev_accept_r_g, evsrv_accept, srv_socket_g, EV_READ);
		ev_io_start(l_g, &ev_accept_r_g);
		Log(SERVERLOG, "  %s - evserver child: re-spawned [pid: %d]\r\n", timestamp, getpid());
		(*printer)("  %s - evserver child: re-spawned [pid: %d]\r\n", timestamp, getpid());
		if (forkcount > 1){ 
            sprintf(serverLogfileName, (char*)"%s/serverlog%d-%d.txt", logsfolder, port_g, getpid());
            sprintf(dbTimeLogfileName, (char*)"%s/dbtimelog%d-%d.txt", logsfolder, port_g, getpid());
        // need the new fork to connect to a database
        Restart();
// init server
int evsrv_init(const string &interfaceKind, int port, char* arg) {
    if (srv_socket_g != -1) 
        ReportBug((char*)"INFO: evserver: server already initialized\r\n");
        return -1;
    if (arg) {
        // parse additional arguments
        // cut evsrv:
        string args(arg + 6);
        // parms are split by ','
        const char *s = args.c_str();
        const char *e = strchr(s, ',');
        while (true) {
            string command(s);
            if (e)  command = string(s, e - s);
            // now split arg/value
            const char *eq = strchr(s, '=');
            if (eq) {
                string option(s, eq - s);
                string val(eq + 1);
                if (e)  val = string(eq + 1, e - eq - 1);
                if (option == "fork") {
					no_children_g = atoi(val.c_str());
                    if (no_children_g > MAX_CHILDREN_D)  no_children_g = MAX_CHILDREN_D;
			else  ReportBug((char*)"Invalid argument to evserver: '%s'\r\n", command.c_str());
            if (e) 
                if (!*s) break;
                e = strchr(e + 1, ',');
			else break;
    interface_g = interfaceKind;
    port_g = port;
    // uf, got socket, now do EV
    l_g = EV_DEFAULT;
#ifdef WIN32
	if (InitWinSock() == FAILRULE_BIT) 
        Log(SERVERLOG, "evsrv_init: WSAStartup failed\r\n");
        return -1;
    srv_socket_g = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (srv_socket_g < 0) {
        Log(SERVERLOG, "evsrv_init: socket() failed, errno: %s\r\n", strerror(errno));
        return -1;
	int on = 1;
#ifdef WIN32
	setsockopt(srv_socket_g, SOL_SOCKET, SO_REUSEADDR, (char*) &on, sizeof(on));
	setsockopt(srv_socket_g, SOL_SOCKET, SO_REUSEADDR, (void*) &on, sizeof(on));
    // non blocking
    if (setnonblocking(srv_socket_g) == -1)  
		Log(SERVERLOG, "evsrv_init: non-blocking ending init\r\n");
 		return -1;
	// bind the socket to its port
	sockaddr_in localAddr;
	memset(&localAddr, 0, sizeof(localAddr));
	localAddr.sin_family = AF_INET;
    if (!inet_aton(interface_g.c_str(), &localAddr.sin_addr)) {
        Log(SERVERLOG, "evsrv_init: inet_aton failed, errno: %s\r\n", strerror(errno));
        return -1;
	localAddr.sin_port = htons(port_g);
	if (::bind(srv_socket_g, (sockaddr *) &localAddr, sizeof(sockaddr_in)) < 0) 
        (*printer)("ERROR: Bind socket attempt failed: port(%d). CS server already running? \r\n", port_g); // : %d\r\n",getpid())
        return -1; // typical when server is already running and cron tries to start
#ifdef EVSERVER_FORK
    int parent_after_fork = -1;
	Log(SERVERLOG, "evserver child fork requests: %d\r\n", no_children_g);
    if (no_children_g > 0) {
        cur_children_g = 0;
        // fork
        int forked = 0;
        for (int i = 0; i < no_children_g; i++) {
            forked = fork_child();
            if (forked == -1) { // failed
                // continue by ourself
                parent_after_fork = 1;
                break;
            } else if (forked == 1) {
                parent_after_fork = 0;
				int oldserverlog = serverLog;
				serverLog = FILE_LOG;
                Log(SERVERLOG, "  evserver: child %d alive\r\n", getpid());
				serverLog = oldserverlog;
                break;
			else  parent_after_fork = 1;
	if (parent_after_fork == 1 && cur_children_g > 0) return 1; // parent of child does not accept/handle connections
    if (listen(srv_socket_g, listen_queue_length_g) < 0) {
        Log(SERVERLOG, "evserver: listen() failed, errno: %s\r\n", strerror(errno));
        return -1;
    // test timer
    ev_timer_init(&tt_g, timeout_cb, 5, 5);
    ev_timer_start(l_g, &tt_g);
    // socket listener
    ev_io_init(&ev_accept_r_g, evsrv_accept, srv_socket_g, EV_READ);
    ev_io_start(l_g, &ev_accept_r_g);
	Log(SERVERLOG, "  evserver: running pid: %d\r\n",getpid());
    (*printer)("  evserver: running pid: %d\r\n", getpid());
    if (forkcount > 1) {
        sprintf(serverLogfileName, (char*)"%s/serverlog%d-%d.txt", logsfolder, port, getpid());
        sprintf(dbTimeLogfileName, (char*)"%s/dbtimelog%d-%d.txt", logsfolder, port_g, getpid());
    return 1;
// starts main server loop
int evsrv_run()
    if (!l_g) 
        ReportBug((char*)"evsrv_run() called with no ev loop initialized");
        (*printer)((char*)"no ev loop initialized, nothing to do\r\n");
        return -1;
    if (parent_g) Log(SERVERLOG, "evserver: parent ready (pid = %d), fork=%d\r\n", getpid(), no_children_g);
	else Log(SERVERLOG, "  evserver: child ready (pid = %d)\r\n", getpid());
	if (!parent_g) (*printer)((char*)"EVServer ready: pid: %d %s\r\n",getpid(),serverLogfileName);
    while (true) ev_run(l_g, 0);
    return 1;
// handler for incomming connections
static void evsrv_accept(EV_P_ ev_io *w, int revents)
    int fd = -1;
    struct sockaddr_storage addr;
    socklen_t sock_len = sizeof(addr);
    // try to accept as many connections as possible
    while (true)
        fd = accept(w->fd, (struct sockaddr *)&(addr), &sock_len);
        if (fd == -1) {
            if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) return; // cannot accept at this time, ev will call us again later
            if (errno == EMFILE || errno == ENFILE)  return; // per-process/system table full, should be freed shortly, just do nothing for now
            Log(SERVERLOG, "evserver: error accepting connection, errno: %s", strerror(errno));
            return;
        if (settcpnodelay(fd) == -1) {
            Log(SERVERLOG, "evserver: could not set TCP_NODELAY, errno: %s", strerror(errno));
            return;
        if (setnonblocking(fd) == -1 || setnonblocking(fd) == -1)  return;
        new Client_t(fd, l_g, ++qsize);
        ++ev_pending;
        if (ev_pending > ev_max) ev_max = ev_pending;
static void client_read(EV_P_ ev_io *w, int revents)
    Client_t *client = (Client_t*)w->data;
    int r = client->recv_data();
    if (r < 0) {
        if (errno == EAGAIN)  return;
        ReportBug( "evserver: got error on read (errno: %s) dropping client %d\r\n", strerror(errno), w->fd);
        client-> Qdown();
        --qsize;
        return;
    else if (r == 0) {
        // client closed connection, lets close ours
        client->Qdown();
        --qsize;
        return;
    // ok, got some data, do we have complete request?
    r = client->received_request();
    if (r == 0)  return; // no, read some more data
    if (r < 0) {
        // invalid request
        ReportBug( "evserver: received invalid request from %d, ignoring\r\n", w->fd);
        client->Qdown();
        --qsize;
        return;
    // yes, we have, do the chat
    r = evsrv_do_chat(client);
    if (r < 0) {
        // could not process it
        client->Qdown();
        --qsize;
        return;
    r = client->send_data();
	if (client->data)
		myfree(client->data);
		client->data = NULL;
	if (r == 0) return; // it means there is still some data to be sent, which will be done next time this watcher is woken - up by ev_loop
	if (r < 0) ReportBug( "evserver: could not sent data to client: %d\r\n", client->fd);
	// else if (r == 1)  // client handling finished
    client->Qdown();
    --qsize;
static void client_write(EV_P_ ev_io *w, int revents)
    Client_t *client = (Client_t*)w->data;
    int r = client->send_data();
    if (r < 0) {
        ReportBug( "evserver: could not sent data to client: %d\r\n", client->fd);
        client->Qdown();
        return;
    if (r == 1) // finished output
        client->Qdown();
    // if r = 0, it means there is still some data to be sent, which will be done next time this watcher is woken-up by ev_loop
int evsrv_do_chat(Client_t *client)
 	uint64 starttime = ElapsedMilliseconds(); 
    client->prepare_for_chat();
    client->startnltime = starttime; // began cs processing nL here
    echo = false;
	bool restarted = false;
	if (!client->data) 	client->data = (char*) mymalloc(outputsize+8);
    if (!client->data)
        (*printer)("Malloc failed for child data\r\n");
        if (!client->data) ReportBug("FATAL: Child malloc failed");
RESTART_RETRY:
    cs_qsize = client->q;
    callStartTime = client->starttime;
	int turn = PerformChat(
        client->user,
        client->bot,
        client->message, // input
        (char*)client->ip.c_str(),
        client->data); // where output goes
    client->endnltime = ElapsedMilliseconds();
	if (turn == PENDING_RESTART) // do user over again in a moment
		restarted = true;
		Log(SERVERLOG,"Restart Request: pid: %d %s \r\n",getpid(),client->user);
		Restart();
		*client->data = 0;
		char* at = SkipWhitespace(client->message);
		if (*at != ':')	goto RESTART_RETRY;
		strcpy(client->data,"Restarted");
	if (!*client->data) 
		client->data[0] = ' ';
		client->data[1] = 0;
		client->data[2] = 0xfe; //ctrl markers
		client->data[3] = 0xff;
		client->data[4] = 0;	// null terminate hidden why data after room for positive ctrlz
#endif /* EVSERVER */
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

evserver.cpp

Latest commit

History

evserver.cpp

File metadata and controls
