---

dataserver.cc

Go to the documentation of this file.

//**************************************************************************
#include <ctype.h>
#include <iomanip.h>
#include "errcode.h"
#include "misc.h"
#include "myendian.h"
#include "dataserver.h"
#include "mythread.h"
//**************************************************************************
DEFINE_TEMPLATE_THREAD_CLASS(DataServerMainThread, DataServer);
DEFINE_TEMPLATE_THREAD_CLASS(DataServerChildThread, ChildThreadStruct);
//**************************************************************************
DataServer::DataServer()
{
Parent = new DataServerMainThread;
keytable = new KeyTable;
callback = NULL;
p_run_flag = 0;
}
//**************************************************************************
DataServer::DataServer(int port)
{
Parent = new DataServerMainThread;
keytable = new KeyTable;
callback = NULL;
p_run_flag = 0;
Start(port);
}
//**************************************************************************
DataServer::~DataServer()
{
if (p_run_flag)
        Terminate();
delete Parent;
delete keytable;
}
//**************************************************************************
void DataServer::RegisterCallback(void (*cb)(char*))
{
callback = cb;
}
//**************************************************************************
int DataServer::NumPendingThreads()
{
int num_pendingthreads = 0;
int size = child.size();

for (int i = 0; i < size; i++)
        {
         if (child[i]->isRunning)
                ++num_pendingthreads;
        }

return num_pendingthreads;
}
//**************************************************************************
int DataServer::Terminate()
{
Stop();
Finalize();
p_run_flag = 0;
cout << "----------------------------------------------------------------------" << endl;
cout << "Stopping DataServer (ID = " << p_serverID << ")." << endl;
cout << "----------------------------------------------------------------------" << endl;
cout << "Total Bytes sent = " << setprecision(5) << BytesSent() << " at rate of " << setprecision(4) << SendRate() << " Mbps." << endl;
cout << "Total Bytes recd = " << setprecision(5) << BytesRecd() << " at rate of " << setprecision(4) << RecvRate() << " Mbps." << endl;
cout << "Total Uptime = " << myTimer()-start_time << " ms." << endl;
cout << "----------------------------------------------------------------------" << endl;
cout.flush();

return 0;
}
//**************************************************************************
int DataServer::Stop()
{
int n = NumPendingThreads();

if (n)
        {
         cout << "DataServer: Waiting for " << n << " client(s) to exit...." << endl << flush;
        }

while (NumPendingThreads())
        {
         mySleep(1000); // Sleep for 1 ms
        }

return 0;
}
//**************************************************************************
int DataServer::Start(int port)
{
Register();     // Register all the variables
static int count = 0;
p_serverID = ++count;
p_run_flag = 1;

int numtries = 1;
while (!TCPSocket::Bind(port)) 
        {
         cerr << "Retrying after 5 seconds." << endl << flush;
         mySleep(5*1000000);    // Sleep for 5 seconds...
         if (numtries++ == 20)
                return POSSE_FAILURE;
        }

char hostname[200];
gethostname(hostname,200);
cout << "DataServer: Binding for " << hostname << ":" << port << endl << flush;

TCPSocket::Listen();

cout << "----------------------------------------------------------------------" << endl;
cout << "Starting DataServer (ID = " << p_serverID << ")" << endl;
cout << "----------------------------------------------------------------------" << endl << flush;
Parent->Start(this);

start_time = myTimer();

return POSSE_SUCCESS;
}
//**************************************************************************
int DataServer::Finalize()
{
TCPSocket::InboundClose();
Parent->Kill();

int size = child.size();
for (int i = 0; i < size; i++)
        {
         delete child[i]->client;
         delete child[i]->th;
         delete child[i];
        }
child.clear();
p_hash_table.clear();

return 0;
}
//**************************************************************************
int DataServer::AcceptClient()
{
RemoteSocket *Client = new RemoteSocket;
*((TCPSocket*) Client) = *((TCPSocket*) this);

ChildThreadStruct *T = new ChildThreadStruct;
T->isRunning = 0;
T->server = this; 
T->ID = child.size();
T->th = NULL;
T->client = Client;
child.push_back(T);

int fail = 0;
while (!Client->Accept())       // Blocks until a client connection is received
        {
         fail = 1;
         cerr << "Retrying connection after 2 seconds." << endl << flush;
         mySleep(2*1000000);    // Wait for 2 sec
        }

if (fail)
        {
         cerr << "Connected." << endl << flush;
        }

return T->ID;
}
//**************************************************************************
void DataServer::RunThread(int ID)
{
int count = 0;
int verification_number = VERIFICATION_NUMBER;
int endiantype = isBigEndian();
RemoteSocket *C = child[ID]->client;
char clientname[200];

cout << "======================================================================" << endl;
cout << "Accepting connection from client (ID = " << ID+1 << ") : " << flush;
//C->RecvLine(clientname);
GetKeyword(C, clientname);
cout << clientname << ":" << C->PeerPort() << endl;
if (strcmp(C->PeerName(), clientname) != 0)
        {
         cout << "\t[Tunneling via " << C->PeerName() << "]" << endl << flush;
        }
cout << "======================================================================" << endl;
cout.flush();
char keyword[100];

strcpy(keyword, "");
int error_flag = 0;

while (strcmp(keyword, "_q_private") != 0)
        {
         GetKeyword(C, keyword);  // Blocking call, does not return till the client sends a keyword
         error_flag = 1;
#ifdef ANIDEBUG
         cout << "Got keyword (from client " << ID << "): " << keyword << endl << flush;
#endif

         if (strcmp(keyword, "_q_private") == 0)
                error_flag = 0;

         Register_Variable_Internal(C, error_flag, keyword, "_start_private", "ro", verification_number);
         Register_Variable_Internal(C, error_flag, keyword, "_isBigEndian", "ro", endiantype);

         if (!strcmp(keyword, "Rinternal_ftp")) // Recv File from Client
                {
                 char filename[200];
                 error_flag = 0;
                 C->RecvLine(filename);
                 C->RecvFile(filename);
                }
         if (!strcmp(keyword, "internal_ftp")) // Send File Requested to Client
                {
                 char filename[200];
                 error_flag = 0; 
                 C->RecvLine(filename);
                 C->SendFile(filename);
                }

         Register(keyword, C, error_flag);
                 
         if (!strcmp(keyword, "keytable")) 
                {
                 Wait();
                        UpdateKeyTable();
                 Post();
                }

         Register_Structure_Internal(C, error_flag, keyword, "keytable", "ro", *keytable);

         if (error_flag == 1)
                {
                 cout << "Unidentified keyword \"" << keyword << "\"!!" << endl << flush;
                 SendVar<int>(0);
                }

         if (error_flag == 2 && !isRunningMPI()) // Updating of "rw" variable
                {
                 char newkeyword[100];
                 strcpy(newkeyword, &keyword[1]);
                 if (callback)
                        callback(newkeyword);
                 if (keytable->hasCallback(newkeyword))
                        keytable->Callback(newkeyword);
                }

         ++count;
        }
cout << "======================================================================" << endl;
cout << "Releasing client (ID = " << ID+1 << "): " << clientname << ":" << C->PeerPort() << endl;
cout << "======================================================================" << endl;
cout.flush();

return;
}
//**************************************************************************
int DataServer::RegisterKey(char *key, DataType desc, char *vartype, int elem_size, int num_elems, char *perm, void *ptr)
{
Hash_Table::const_iterator iter = p_hash_table.find(key);
if (iter != p_hash_table.end())
        {
         cout << "DataServer::RegisterKey: Key \"" << key << "\" already registered!!" << endl << flush;
         return iter->second;   // Key already registered
        }

char type[80];
int n = 0;

for (int i = 0; i < strlen(vartype); i++)
        {
         if (!isspace(vartype[i]))
                type[n++] = vartype[i];
        }
type[n] = '\0';

int id = keytable->n;
++keytable->n;
p_hash_table[key] = id;

#ifdef ANIDEBUG
cout << "Registering key[" << id+1 << "] = \"" << key << "\"" << endl << flush;
#endif

keytable->Add();
keytable->key[id].sem = new Semaphore;
keytable->key[id].id = id;
keytable->key[id].is_locked = 0;
keytable->key[id].is_dirty = 0;
keytable->key[id].elem_size = elem_size;
keytable->key[id].num_elems = num_elems;
keytable->key[id].ptr = ptr;
if (strcmp(perm, "rw") == 0)
        keytable->key[id].perm = KEY_READ_WRITE;
else if (strcmp(perm, "ro") == 0)
        keytable->key[id].perm = KEY_READ_ONLY;
else
        {
         cout << "Invalid permission \"" << perm << "\" for keyword \"" << key << "\"!!" << endl << flush;
         exit(-1);
        }
strcpy(keytable->key[id].key, key);
keytable->key[id].description = desc;
strcpy(keytable->key[id].type, type);

keytable->key[id].vartype = UNKNOWN;

static char typedesc[][20] = {"char", "short", "int", "long", "float", "double"};
static VariableType vtype[] = {CHAR, SHORT, INT, LONG, FLOAT, DOUBLE};
static int ntype = sizeof(typedesc)/sizeof(char[20]);

for (int j = 0; j < ntype; j++)
        {
         if (strcmp(type, typedesc[j]) == 0)
                { 
                 keytable->key[id].vartype = vtype[j];
                 break; 
                }
        }

return id;
}
//**************************************************************************
int DataServer::UpdateKeyTable()
{
KeyTable *K = keytable;
for (int i = 0; i < K->n; i++)
        {
         K->key[i].is_locked = K->key[i].sem->WaitCount() - K->key[i].sem->PostCount();
         int maxlen = sizeof(K->key[i].value);
         memcpy(K->key[i].value, GetValue(K->key[i].key), maxlen-1);
         K->key[i].value[maxlen-1] = '\0';
         if (K->key[i].is_locked < 0) K->key[i].is_locked = 0;
        }
return 0;
}
//**************************************************************************
int DataServer::ClearLocks()
{
int n = 0;
for (int i = 0; i < keytable->n; i++)
        {
         keytable->key[i].is_locked = keytable->key[i].sem->WaitCount() - keytable->key[i].sem->PostCount();
         if (keytable->key[i].is_locked > 0)
                {
                 ++n;
                 cout << "\tRemoving lock on key \"" << keytable->key[i].key << "\"." << endl << flush;
                 keytable->key[i].sem->Post();

                }
        }
return n;
}
//**************************************************************************
char *DataServer::GetValue(int i)
{
return GetValue(Key(i));
}
//**************************************************************************
char *DataServer::GetValue(char *keyword)
{
char *val = new char[100];
VariableType vtype = UNKNOWN;
KeyTable *K = keytable;

int handle = K->getKeyHandle(keyword);
if (handle < K->n && K->isViewableVar(handle))
        vtype = K->key[handle].vartype;

// If ptr points to memory under 1 MB, it is probably invalid!!
if (vtype != UNKNOWN && (int) K->key[handle].ptr < 1024*1024)
        vtype = UNKNOWN;

switch (vtype)
        {
         case CHAR:
                if (K->isCharArray(handle))
                        {
                         memcpy(val, K->key[handle].ptr, 99);
                         val[99] = '\0';
                        }
                else
                        {
                         char x_char;
                         memcpy(&x_char, K->key[handle].ptr, sizeof(char));
                         sprintf(val, "%c\0", x_char);
                        }
                break;

         case SHORT:
                short x_short;
                memcpy(&x_short, K->key[handle].ptr, sizeof(short));
                sprintf(val, "%d\0", x_short);
                break;

         case INT:
                int x_int;
                memcpy(&x_int, K->key[handle].ptr, sizeof(int));
                sprintf(val, "%d\0", x_int);
                break;

         case LONG:
                long x_long;
                memcpy(&x_long, K->key[handle].ptr, sizeof(long));
                sprintf(val, "%ld\0", x_long);
                break;

         case FLOAT:
                float x_float;
                memcpy(&x_float, K->key[handle].ptr, sizeof(float));
                sprintf(val, "%g\0", x_float);
                break;

         case DOUBLE:
                double x_double;
                memcpy(&x_double, K->key[handle].ptr, sizeof(double));
                sprintf(val, "%g\0", x_double);
                break;

         default:       // case UNKNOWN:
                strcpy(val, "error");
                break;
        }
        
return val;
}
//**************************************************************************
void DataServer::PrintDirty(ostream &out, int change)
{
for (int i = 0; i < NumKeys(); i++)
        if (IsDirtyID(i, change))
                out << "\tDataServer: New value for \"" << Key(i) << "\" = " << GetValue(i) << endl << flush;
}
//**************************************************************************
void DataServerMainThread::Run(DataServer *S)
{
signal(SIGABRT, SIG_IGN);
while (!isExit())
        {
         int id = S->AcceptClient(); // Blocking call, does not return till
                                     // a new client request is received.
         S->child[id]->th = new DataServerChildThread;
         S->child[id]->th->Start(S->child[id]);
        }
}
//**************************************************************************
void DataServerChildThread::Run(ChildThreadStruct *T)
{
trap_sigpipe();

try
        {
         T->isRunning = 1;
         T->server->RunThread(T->ID);
         T->isRunning = 0;
        }
catch (int &e)
        {
#ifndef __CYGWIN32__
         if (e == ERR_BAD_SOCKET || e == ERR_RECV_MESG || e == ERR_SEND_MESG)
                {
                 cout << "DataServer: Killing child thread " << T->th->ID() << "." << endl;
                }
         if (e == ERR_SIGPIPE)
                {
                 cout << "DataServer:SIGPIPE: Killing child thread " << T->th->ID() << "." << endl;
                }
#endif
         if (e == ERR_BAD_DATA)
                {
                 return;
                }
         T->server->ClearLocks();
         T->isRunning = 0;
         T->th->Kill();
        }
catch (...)
        {
         cout << "Another exception!!" << endl;
        }
}
//**************************************************************************
int DataServer::Synchronize(char *filename)
{
if (filename)
        {
         ofstream file(filename, ofstream::app);
         PrintDirty(file, 1);
         file.close();
        }
else
        PrintDirty(cout, 1);

return POSSE_SUCCESS;
}
//**************************************************************************
int DataServer::Register(char *keyword, RemoteSocket *client, int &error_flag, bool register_flag)
{
if (RegisterInternal(keyword, client, error_flag, register_flag) == 1)
        return 1;
return RegisterExternal(keyword, client, error_flag, register_flag);
}
//**************************************************************************

---