// -*- c++ -*- //------------------------------------------------------------------------------ // $Id: bufio_testc.cpp,v 1.9 2006/07/20 02:30:55 vlg Exp $ //------------------------------------------------------------------------------ // This program is free software; you can redistribute it and/or // modify it under the terms of the GNU General Public License // as published by the Free Software Foundation; either version // 2 of the License, or (at your option) any later version. // // Date: April 12, 1999 //------------------------------------------------------------------------------ const char title[] = " NAME \n" " \n" " bufio_testc - [buf]fered [io] [testc]lient program is \n" " the client side that tests Socketbuf class functionality. \n" " \n" " SYNOPSIS \n" " \n" " shell> bufio_testc --port port_name@host_name \n" " [--log-file logfile] [--mask mask] [--daemon] \n" " [--build-dir STRING] \n" " \n" " OPTIONS: \n" " \n" " --build-dir STRING - Directory where executables are located. \n" " \n" " --port $ASSAPORT[@localhost] \n" " \n" " If host_name is 'localhost' or hostname of the local machine, \n" " bufio_testc will span off bufio_tests server as: \n" " \n" " % bufio_tests --port port_name --log-file=bufioS.log \n" " \n" " If not, bufio_tests server is assumed to be running on \n" " the 'host_name' computer. \n" " \n" " --log-file FILE - Message logging file \n" " --mask MASK - Message logging mask \n" " --daemon - Run as daemon process \n" " \n" " EXAMPLES \n" " \n" " shell> bufio_testc --port=assatest --log-file=bufioC.log \n" " --mask=0x40010001 \n" " \n" " DESCRIPTION \n" " \n" " This test program spans off server-side (if host_name indicates local \n" " host or absent) with configuration to listen on local port 'assatest', \n" " connects to it and then performs basic tests on buffering IO. \n" " \n" " Three tests are performed: \n" " \n" " 1) 1 byte is sent, received and must match. \n" " \n" " 2) A short data segment is sent, received and must match. \n" " \n" " 3) A data segment larger that Socketbuf internal buffer size \n" " (but less then 2832 bytes) is sent, received and must match. \n" " \n" " 4) Send signal USR1 to the server side. This will swicth \n" " it to 'delay mode.' \n" " Send 20 bytes of data. The first four bytes are reflected \n" " right away. The remaining 16 bytes will be sent back one \n" " second later. \n" " Receive data with 2 xdrIOBuffers: 12 bytes long (header) and \n" " 8 bytes long (body and tail). \n" " \n" " SEE ALSO \n" " + Server side, bufio_tests.cpp for further details. \n" " + echoxdr_testc / echoxdr test pair for XDR-encoded basic type \n" " trasfer by ASSA library. \n" " \n" " AUTHOR \n" " Vladislav Grinchenko \n" " \n"; //------------------------------------------------------------------------------ #if !defined (WIN32) #include using std::string; #include #include using namespace std; #include // stat(2) #include // stat(2) //Assa Includes #include "assa/Logger.h" #include "assa/Assure.h" #include "assa/GenServer.h" #include "assa/Singleton.h" #include "assa/ServiceHandler.h" #include "assa/Reactor.h" #include "assa/IPv4Socket.h" #include "assa/Connector.h" #include "assa/INETAddress.h" #include "assa/xdrIOBuffer.h" #include "assa/CommonUtils.h" using namespace ASSA; char t2_resp[] = "Vladislav Grinchenko"; //------------------------------------------------------------------------------ // SvrProxy //------------------------------------------------------------------------------ class SvrProxy : public ServiceHandler { public: enum { closed, opened, t1_waiting, t2_waiting, t3_waiting, t4_waiting }; SvrProxy (); ~SvrProxy (); int open (); void close (); private: int handle_read (int fd_); int handle_timeout (TimerId tid_); void run_test1 (); void run_test2 (); void run_test3 (); void run_test4 (); TimerId m_tid; // Timer ID TimeVal m_timeout; // IO operations timeout Reactor& m_r; // Reference to Reactor int m_state; // Object's state enum { FMAXSIZE = 2832 }; // Max frame size in bytes char m_t3_data[FMAXSIZE]; xdrIOBuffer* m_buf; xdrIOBuffer* m_buf2; }; //------------------------------------------------------------------------------ // Client //------------------------------------------------------------------------------ class Client : public GenServer, public Singleton { public: Client (); ~Client (); void init_service(); void process_events(); void send_usr1_sig () { Assure_exit (kill (m_server_pid, SIGUSR1) == 0); } string get_build_dir () const { return m_build_dir; } private: SvrProxy* m_svr; // Server proxy INETAddress* m_svr_addr; // Where server lives pid_t m_server_pid; string m_build_dir; Connector m_conn; }; //------------------------------------------------------------------------------ Client::Client () { trace("Client::Client"); ::unlink ("bufio_test.dat"); add_opt (0, "build-dir", &m_build_dir); m_log_file = "bufioC.log"; } Client::~Client () { trace("Client::~Client"); if (m_svr != 0) { delete m_svr; } if (m_svr_addr != 0) { delete m_svr_addr; } } void Client::init_service () { trace("Client::init_service"); if (m_build_dir.length () == 0) { m_build_dir = ASSA::Utils::get_cwd_name (); } DL((APP,"build-dir = \"%s\"\n", m_build_dir.c_str ())); m_svr_addr = new INETAddress (get_port ().c_str ()); m_svr_addr->dump (); Assure_exit (m_svr_addr->good ()); /*--- If local host then span off the server side. ---*/ string addr_hn = m_svr_addr->getHostName (); char hn [64]; Assure_exit (::gethostname (hn, sizeof (hn)-1) == 0); struct hostent* hentry; hentry = ::gethostbyname (hn); string hostname (hentry->h_name); string svr_name ("bufio_tests"); DL((TRACE,"Requested host name: %s\n", addr_hn.c_str ())); DL((TRACE,"Local host name: %s\n", hostname.c_str ())); if (addr_hn == hostname || addr_hn == string ("localhost") || addr_hn.empty ()) { std::string exec_name (get_build_dir () + "/" + svr_name); Fork ss (Fork::KILL_ON_EXIT, Fork::IGNORE_STATUS); if (ss.isChild ()) { execlp (exec_name.c_str (), exec_name.c_str (), "--port", get_port ().c_str(), "--log-file", "bufioS.log", NULL); // unreachable cerr << "error: " << svr_name << ": "; switch (errno) { case EACCES: cerr << "Permission denied"; break; case ENOEXEC: cerr << "Exec format error"; break; case ENOENT: cerr << "No such file or directory"; break; default: cerr << "errno - " << errno; } cerr << endl; exit (1); } else if (ss.isParent ()) { m_server_pid = ss.getChildPID (); sleep (2); // Give the server a chance to run } } } void Client::process_events() { trace("Client::process_events"); m_svr = new SvrProxy; // Connect to the server TimeVal tout (5.0); if (m_conn.open (tout, ASSA::async, &m_reactor) < 0) { cerr << "Open to "<< get_port() << "@localhost failed\n"; stop_service (); return; } if (m_conn.connect (m_svr, *m_svr_addr) < 0) { cerr << "Connect failed. errno: " << errno << endl; stop_service (); return; } while (service_is_active ()) { m_reactor.waitForEvents (); } } ASSA_DECL_SINGLETON(Client); //------------------------------------------------------------------------------ // SvrProxy's methods //------------------------------------------------------------------------------ SvrProxy::SvrProxy () : ServiceHandler (new IPv4Socket), m_tid (0), m_timeout (4.0), m_r (* Client::get_instance()->get_reactor()), m_state (closed), m_buf (0), m_buf2(0) { trace("SvrProxy::SvrProxy"); memset (m_t3_data, 0, sizeof (m_t3_data)); } SvrProxy::~SvrProxy () { trace("SvrProxy::~SvrProxy"); if (m_buf) { delete m_buf; } if (m_buf2) { delete m_buf2; } } void SvrProxy::close () { trace("SvrProxy::close"); Client::get_instance()->stop_service (); } int SvrProxy::open () { trace("SvrProxy::open"); string msg = "CONNECTION ESTABLISHED"; /*--- We have been called by Connector - connection established. ---*/ DL((TRACE,"%s\n", msg.c_str ())); cout << msg << "\n"; m_state = opened; /*--- Switch socket to non-blocking mode. ---*/ this->get_stream ().turnOptionOn (Socket::nonblocking); this->get_stream ().dumpState (); run_test1 (); return 0; } int SvrProxy::handle_read (int fd_) { trace("SvrProxy::handle_read"); IPv4Socket& s = *this; if (s.getHandler () != fd_) return -1; if (m_state == t1_waiting) { char t1 = 'V'; char t1_resp = 'v'; Assure_exit (s.read (&t1_resp, 1) == 1 && t1 == t1_resp); DL((TRACE,"T1: OK!\n")); cout << "T1: OK\n" << flush; run_test2 (); // Go to T2 } else if (m_state == t2_waiting) { s >> *m_buf; if (! *m_buf) { // Wait for entire message to accumulate. return 0; } /* Compare data */ Assure_exit (memcmp (t2_resp, m_buf->str (), m_buf->size ()) == 0); delete m_buf; m_buf = 0; DL((TRACE,"T2: OK!\n")); cout << "T2: OK\n" << flush; run_test3 (); // Go to T3 } else if (m_state == t3_waiting) { s >> *m_buf; if (! *m_buf) { return 0; } Assure_exit (memcmp (m_t3_data, m_buf->str (), m_buf->size ()) == 0); delete m_buf; m_buf = 0; DL((TRACE,"T3: OK!\n")); cout << "T3: OK\n" << flush; run_test4 (); // Go to T4 } else if (m_state == t4_waiting) { /* First stage: receive first 12-bytes frame */ if (m_buf2 == 0) { s >> *m_buf; if (! *m_buf) return 0; /* Wait for the whole thing */ DL((TRACE,"Received first 12 bytes\n")); delete m_buf; m_buf = 0; /* First 12-bytes frame received */ m_buf2 = new xdrIOBuffer (8); } /* Second stage: receive second 8-bytes frame. Note here, reading remainging 8 bytes of the first stage will bring entier 16 bytes into the Socketbuf buffer. Reactor's select() won't flag the descriptor as ready for reading. So, we read and possibly fail before we call Reactor::waitForEvents (). */ s >> *m_buf2; if (! *m_buf2) return 0; /* Wait for the whole thing */ DL((TRACE,"Received remaining 8 bytes\n")); delete m_buf2; m_buf2 = 0; DL((TRACE,"T4: OK!\n")); cout << "T4: OK\n" << flush; close (); // End the test } return BYTES_LEFT_IN_SOCKBUF(s); } int SvrProxy::handle_timeout (TimerId /* tid_ */) { trace("SvrProxy::handle_timeout"); string msg; /* Remove IO Handler callback */ Assure_exit (m_r.removeHandler (this, READ_EVENT)); string tn; if (m_state == t1_waiting) tn = "T1"; else if (m_state == t2_waiting) tn = "T2"; else if (m_state == t3_waiting) tn = "T3"; else if (m_state == t4_waiting) tn = "T4"; else tn = "UNKNOWN STATE"; msg = "Timed out waiting on " + tn + " response."; EL((ASSA::ASSAERR,"%s\n", msg.c_str())); cout << msg << "\n"; close (); return -1; // Remove Timer Handler } void SvrProxy::run_test1 () { trace("SvrProxy::run_test1"); char c = 'V'; IPv4Socket& s = *this; DL((TRACE,"===================================\n")); DL((TRACE,"= Test 1 =\n")); DL((TRACE,"===================================\n")); /* Send 1 byte of data. */ Assure_exit (s.write (&c, 1) == 1); s << flush; m_r.registerIOHandler (this, s.getHandler (), READ_EVENT); m_tid = m_r.registerTimerHandler (this, m_timeout); m_state = t1_waiting; } void SvrProxy::run_test2 () { trace("SvrProxy::run_test2"); char buf[] = "Vladislav Grinchenko"; int len = strlen(buf); IPv4Socket& s = *this; DL((TRACE,"===================================\n")); DL((TRACE,"= Test 2 =\n")); DL((TRACE,"===================================\n")); /* Remove Test1 timer */ if (m_tid) { Assure_exit (m_r.removeTimerHandler (m_tid)); m_tid = 0; } Assure_exit (s.write (buf, len) == len); s << flush; m_tid = m_r.registerTimerHandler (this, m_timeout); m_state = t2_waiting; m_buf = new xdrIOBuffer (len); } void SvrProxy::run_test3 () { trace("SvrProxy::run_test3"); char fname[] = "bufio_test.dat"; register char c; DL((TRACE,"===================================\n")); DL((TRACE,"= Test 3 =\n")); DL((TRACE,"===================================\n")); /* Remove Test1 timer */ if (m_tid) { Assure_exit (m_r.removeTimerHandler (m_tid)); m_tid = 0; } /* Test if input buffer is available. */ struct stat* fstat; if (stat (fname, fstat) == 0) { if (S_ISREG (fstat->st_mode)) { if (fstat->st_size > FMAXSIZE) { cerr << "Input file \"" << fname << '\"' << " cannot be larger than " << FMAXSIZE << " bytes\n"; Assure_exit(false); } } else { cerr << fname << " is not a regular file\n\n"; Assure_exit(false); } } else { // Create a file ofstream dummy (fname); for (int j=0; j < FMAXSIZE; ++j) { dummy << c; } dummy << ends; dummy.close (); } /* Read file that is bigger than single TCP frame can handle. This will force frame fragmentation and I can test if xdrIOBuffer handles it correctly. */ ifstream file (fname); if (! file) { // input will not succeed cerr << "Failed to open file " << fname << endl; Assure_exit(false); } register int len = 0; while (file >> m_t3_data[len++] && len < FMAXSIZE) {} file.close (); m_buf = new xdrIOBuffer (len); IPv4Socket& s = *this; Assure_exit (s.write (m_t3_data, len) == len); s << flush; m_tid = m_r.registerTimerHandler (this, m_timeout); m_state = t3_waiting; } void SvrProxy::run_test4 () { trace("SvrProxy::run_test4"); // 12345678901234567890 : 20 characters char buf[] = "vladgrinchenkomatrix"; DL((TRACE,"===================================\n")); DL((TRACE,"= Test 4 =\n")); DL((TRACE,"===================================\n")); int len = strlen(buf); IPv4Socket& s = *this; /* Remove timer */ if (m_tid) { Assure_exit (m_r.removeTimerHandler (m_tid)); m_tid = 0; } /* Switch server to delay mode */ Client::get_instance ()->send_usr1_sig (); /* Allocate first 12-bytes buffer */ m_buf = new xdrIOBuffer (12); /* Write entire 20 bytes message to the server */ Assure_exit (s.write (buf, len) == len); s << flush; m_tid = m_r.registerTimerHandler (this, m_timeout); m_state = t4_waiting; } //------------------------------------------------------------------------------ // main //------------------------------------------------------------------------------ #define CLIENT Client::get_instance () #endif /* !defined WIN32 */ int main (int argc, char* argv[]) { #if !defined (WIN32) string release ("1.1"); int revision = 0; CLIENT->set_version (release, revision); CLIENT->set_author ("Vladislav Grinchenko"); CLIENT->set_flags (GenServer::RMLOG); CLIENT->init (&argc, argv, title); CLIENT->init_service (); CLIENT->process_events (); #endif /* !defined WIN32 */ return 0; }