/* * $Id: imserver.cxx,v 4.4 2004/05/24 23:45:33 hut66au Exp $ * * Imview, the portable image analysis application * http://www.cmis.csiro.au/Hugues.Talbot/imview * ---------------------------------------------------------- * * Imview is an attempt to provide an image display application * suitable for professional image analysis. It was started in * 1997 and is mostly the result of the efforts of Hugues Talbot, * Image Analysis Project, CSIRO Mathematical and Information * Sciences, with help from others (see the CREDITS files for * more information) * * Imview is Copyrighted (C) 1997-2001 by Hugues Talbot and was * supported in parts by the Australian Commonwealth Science and * Industry Research Organisation. Please see the COPYRIGHT file * for full details. Imview also includes the contributions of * many others. Please see the CREDITS file for full details. * * 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. * * This program 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 General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA. * */ /*------------------------------------------------------------------------ * * Imview as an image server!! * * Uses the async library by Sam Rushing. * * Basically the idea is to be able to embed imview more easily. * * Hugues Talbot 29 Feb 2000 * *-----------------------------------------------------------------------*/ #include "imnmspc.hxx" #include "imview.hxx" #ifdef HAVE_PTHREADS // otherwise don't even bother #include #include // malloc and friends #include #include "imserver.hxx" #include "imageIO.hxx" #include "io/newprefio.hxx" //--------- only one server extern imview_server *is; extern interpreter *imview_interpreter; extern imageIO *IOBlackBox; extern imprefs *fileprefs; extern bool server_terminate; extern bool darwinChild; extern int portfileDescriptor; extern volatile int serverPortNumber; extern volatile bool im_draw_finished; extern volatile bool hasSynced; extern double simple_progress; extern char portfilename[]; imview_server::imview_server() { dco = 0; // no data channel opener in the beginning. } // not sure this is necessary? imview_server::~imview_server() { imview_channel *c; while (!lc.empty()) { c = lc.front(); delete c; lc.pop_front(); } delete dco; // delete the data channel opener dco = 0; } void imview_server::handle_accept (void) { struct sockaddr addr; size_t addr_len = sizeof(sockaddr); imview_channel *jc; srv_dbgprintf("imview server: handle_accept\n"); int fd = dispatcher::accept (&addr, &addr_len); jc = new imview_channel; jc->set_fileno (fd); jc->set_terminator ("\n"); //jc->send(IMPROMPT); // remember it lc.push_back(jc); } datachannel_opener::~datachannel_opener(void) { imview_data *d; srv_dbgprintf("Data channel destructor called\n"); while (!ld.empty()) { d = ld.front(); delete d; ld.pop_front(); } } void datachannel_opener::handle_accept(void) { struct sockaddr addr; size_t addr_len = sizeof(sockaddr); imview_data *jd; srv_dbgprintf("datachannel opener: open_data_channel\n"); int fd = dispatcher::accept(&addr, &addr_len); srv_dbgprintf("New fd for data channel = %d\n", fd); jd = new imview_data; jd->set_fileno(fd); // remember it ld.push_back(jd); return; } // open another socket for the binary data. // int imview_server::open_data_channel(string &s) { int port, endport; int rescreate, resbind, reslisten; struct sockaddr_in addr; port = 0; if (dco == 0) { // create a new one dco = new datachannel_opener; dco->setparent(this); } if (dco->is_closed()) { // reopen the channel dco->reopen_channel(); } if (!dco->is_accepting()) { port = port_start + 1; // data ports interleaved with command ports srv_dbgprintf("Opening data channel\n"); if ((rescreate = dco->create_socket (AF_INET, SOCK_STREAM)) < 0) { s = "Could not create socket: "; s += strerror(errno); return -1; } addr.sin_family = AF_INET; addr.sin_addr.s_addr = htonl (INADDR_ANY); endport = port + 100; while (port < endport) { addr.sin_port = htons (port); if ((resbind = dco->bind ((struct sockaddr *) &addr, sizeof (addr))) < 0) { srv_dbgprintf("Could not bind socket: %s\n", strerror(errno)); } else { srv_dbgprintf("Binding to port: %d\n", port); break; } port += 2; } if (( reslisten = dco->listen (5)) < 0) { s = "Could not listen to socket: "; s += strerror(errno); return -2; } // all went well, remember the port dco->setport(port); } else { // port already exist, is open and accepting connections. port = dco->port(); } // if all went well return port; } // This is a strong action. The closed // channel will not be available for several minutes! (TCP_WAIT) int imview_server::close_data_channel(void) { if (dco) { dco->close(); // this will shutdown the socket //delete dco; //dco = 0; } return 0; } // called by the server thread void imview_server::manage_results(void) { CLI p; // check all the result channels... for (p = lc.begin(); p != lc.end() ; p++) { if (!(*p)->result_isempty()) { // send the result through the socket srv_dbgprintf("sending result throught the socket\n"); (*p)->sendResult(); } } } // called by the GUI thread void imview_server::manage_commands(void) { CLI p; string aCommand, aResult; // check all the command channels... for (p = lc.begin(); p != lc.end() ; p++) { if (!(*p)->command_isempty()) { // evaluate the command aCommand = (*p)->read_command(); imview_interpreter->parseCommand(aCommand, aResult); // we don't write the result yet dbgprintf("Result to be printed: %s\n", aResult.c_str()); (*p)->write_result(aResult); // atomic // that's it... } } return; } //------------ the data channels int imview_data::channels_id(0); imview_data::imview_data(void) { abort = false; identifier = channels_id++; comm_bufsize = IBFSIZE; // initial buffer size... recv_bufsize = IBFSIZE; received = 0; // nothing received yet! expected = 0; recv_buffer = (char *) malloc(recv_bufsize * sizeof(char)); comm_buffer = (char *) malloc(recv_bufsize * sizeof(char)); if ((recv_buffer == 0) || (comm_buffer == 0)) { cerr << "Buffer reception will fail!\n" ; abort = true; // we won't be able to receive anything. } } // // OK, the non-freeing of recv_buffer requires an explanation: // this buffer is passed to the GUI as the buffer that has received the // data. It is copied into the image header and freed when the GUI // closes the image. Any other mechanism requires an extra memcpy. imview_data::~imview_data(void) { /* free(recv_buffer); DON'T free this buffer. The application will do it! */ /* free(comm_buffer); DON'T free this buffer, it gets freed at the end of the transfer! */ return; } void imview_data::handle_connect(void) { srv_dbgprintf("Data channel #%d connection\n", identifier); return; } void imview_data::handle_read(void) { int result; srv_dbgprintf("Data channel #%d reading...\a", identifier); if (!abort) { result = recv(comm_buffer, comm_bufsize); if (result > 0) { srv_dbgprintf("received %d bytes ", result); // put the result at the end of the bits already received... assert(recv_bufsize != 0); // this would be a disaster. if ((result + received) > recv_bufsize) { srv_dbgprintf("Reallocating buffer ", result); if (expected != 0) { // realloc to exactly the needed size recv_buffer = (char *)realloc(recv_buffer, expected * sizeof(char)); recv_bufsize = expected; } else { // realloc to double the size recv_buffer = (char *)realloc(recv_buffer, recv_bufsize * 2 * sizeof(char)); recv_bufsize *= 2; } if (recv_buffer != 0) { srv_dbgprintf("New size = %d ", recv_bufsize); } else { // risky but interesting stderrprintf("Upload realloc failed! not enough memory\n" "Aborting transfer"); abort = true; handle_close(); return ; } } // copy the data to the buffer memcpy(recv_buffer + received, comm_buffer, result); received += result; srv_dbgprintf("moved... ", result); // set progress report if (expected == 0) { if (received >= HEADER_ID_SIZE) { int *id = (int *)recv_buffer; // this call is protected IMAGE_HEADER *img = IOBlackBox->findHeader(*id); if (img != 0) expected = img->expected_size; } srv_dbgprintf("\b"); } // check again: note, expected should NOT be 0 now if (expected != 0) { simple_progress = (100.0 * (double)received)/((double)expected); srv_dbgprintf("progress = %4.1f%%\b", simple_progress); if (received == expected) { srv_dbgprintf("Transfer completed\n"); // last handshake in fact IS necessary for very small buffers dispatcher::send("THE END\n", 8, 0); // close(); // no need to do anything else handle_close(); return; } } // handshake dispatcher::send(HNDSHK, HSLEN, 0); } else { // usually this is OK, really srv_dbgprintf("Recv error: %s\n", strerror(errno)); } } return; } void imview_data::handle_write(void) { srv_dbgprintf("Data channel #%d writing\n", identifier); return; } void imview_data::handle_close(void) { srv_dbgprintf("Data channel #%d closing\n", identifier); srv_dbgprintf("Received: %d bytes\n", received); if (!abort) { srv_dbgprintf("Making callback now\n"); // tell interpreter to do something with the buffer now put::callback(recv_buffer, received); } else { srv_dbgprintf("The upload was aborted for some reason\n"); // realloc the memory to a useable size recv_buffer = (char *)realloc(recv_buffer, IBFSIZE * sizeof(char)); if (recv_buffer != NULL) recv_bufsize = IBFSIZE; else { srv_dbgprintf("\n" "*** Realloc failed when allocating a very small buffer\n" "*** almost surely this indicates a memory management problem\n"); recv_bufsize = 0; // This is sure to create problems later on. } // delete the header... } // freeing the comm_buffer free(comm_buffer); comm_buffer = 0; // closing the channel dispatcher::close(); // will this work? srv_dbgprintf("Channel %d closed\n", identifier); return; } //------------ Many command channels per server // initializes class variable int imview_channel::channels_id(0); imview_channel::imview_channel(void) : async_chat() { identifier = channels_id++; srv_dbgprintf("New channel constructed, id=%d, semaphores allocated\n", identifier); semaphore_init(&access_command); semaphore_init(&access_result); logged_in = false; // no command will work except USER } imview_channel::~imview_channel(void) { srv_dbgprintf("Channel %d destroyed, semaphore deleted\n", identifier); semaphore_destroy(&access_command); semaphore_destroy(&access_result); } void imview_channel::protect_command(void) { semaphore_down(&access_command); return; } void imview_channel::unprotect_command(void) { semaphore_up(&access_command); return; } void imview_channel::write_command(string &c) { protect_command(); commandBuffer.push_back(c); unprotect_command(); return; } string &imview_channel::read_command(void) { // note that cs is an instance variable protect_command(); cs = commandBuffer.front(); commandBuffer.pop_front(); unprotect_command(); return cs; } bool imview_channel::command_isempty(void) { bool res; protect_command(); res = commandBuffer.empty(); unprotect_command(); return res; } void imview_channel::protect_result(void) { semaphore_down(&access_result); return; } void imview_channel::unprotect_result(void) { semaphore_up(&access_result); return; } void imview_channel::write_result(string &c) { protect_result(); resultBuffer.push_back(c); unprotect_result(); return; } string &imview_channel::read_result(void) { protect_result(); rs = resultBuffer.front(); resultBuffer.pop_front(); unprotect_result(); return rs; } bool imview_channel::result_isempty(void) { bool res; protect_result(); res = resultBuffer.empty(); unprotect_result(); return res; } void imview_channel::sendResult(void) { string &c = read_result(); // atomic string sc = c.substr(0,7); // check for log in keywork // check for session-terminating string if (c == "Bye!\n") { srv_dbgprintf("Closing connection %d\n", identifier); logged_in = false; close(); // end of connection. } else if ( sc == "Welcome" ) { logged_in = true; // all commands will go through now srv_dbgprintf("Logged in now\n"); send(c); //send(IMPROMPT); }else { srv_dbgprintf("About to be sent: %s\n", c.c_str()); send(c); //send(IMPROMPT); } return; } void imview_channel::handle_close (void) { srv_dbgprintf("Channel closing\n"); } void imview_channel::collect_incoming_data (const string& data) { int counter = 0; srv_dbgprintf("From: %d, received %s\n", identifier, data.c_str()); while (!hasSynced) { srv_dbgprintf("Waiting for sync\n"); #ifndef WIN32_NOTCYGWIN myusleep(10000); #else _sleep(1); #endif } srv_dbgprintf("Waiting on display\a", counter); while (!im_draw_finished && (++counter < 100)) { #ifndef WIN32_NOTCYGWIN myusleep(1000); #else _sleep(1); #endif } input_buffer.append (data); } void imview_channel::found_terminator (void) { if (input_buffer.length()) { string sc = input_buffer.substr(0,4); // if not logged in, the only command we check for // is `user' if (logged_in || (cmp_nocase(sc, "USER") == 0) || (cmp_nocase(sc, "QUIT") == 0) || (cmp_nocase(sc, "EXIT") == 0)) { // write the string to the common buffer write_command(input_buffer); // this is atomic } else { string as = "deny\r\n"; write_command(as); } // send("Well received: " + input_buffer + "\n"); // send(IMPROMPT); input_buffer.erase(); } } #ifdef WIN32_NOTCYGWIN void init_winsock (void) { WSADATA wd; WSAStartup (MAKEWORD (1,1), &wd); } #endif // This function is meant to be called as a separate thread. void *init_server(int *portnumber) { int port, endport; int rescreate, resbind, reslisten; struct sockaddr_in addr; struct timeval tv; double wp; // create the server! is = new imview_server; #ifdef WIN32_NOTCYGWIN init_winsock(); #endif if (!portnumber) { port = PORTNB; // why not } else { port = *portnumber; } // useful to remember the `prototype' port is->setport(port); if ((rescreate = is->create_socket (AF_INET, SOCK_STREAM)) < 0) { errprintf("Could not create socket: %s", strerror(errno)); return (void*)1; } addr.sin_family = AF_INET; addr.sin_addr.s_addr = htonl (INADDR_ANY); endport = port + 100; while (port < endport) { addr.sin_port = htons (port); if ((resbind = is->bind ((struct sockaddr *) &addr, sizeof (addr))) < 0) { srv_dbgprintf("Could not bind to socket: %s\n", strerror(errno)); } else { srv_dbgprintf("Binding to port: %d\n", port); // report to the command line // the file was opened before the server was started. srv_dbgprintf("Writing port number to a file or a pipe\n"); if (portfileDescriptor >= 0) { // note that the port number is NOT a short here. int nbw = write(portfileDescriptor, &port, sizeof(int)); if (nbw < 0) { srv_dbgprintf("Writing port number failed: %s\n", strerror(errno)); } close(portfileDescriptor); // will flush the file } else if (darwinChild) { // write to standard error write(2, &port, sizeof(int)); } else { srv_dbgprintf("Port: %d\n", port); // just print the port name printf("Port: %d\n", port); } break; } port += 2; } if (( reslisten = is->listen (5)) < 0) { errprintf("Could not listen to socket: %s", strerror(errno)); } srv_dbgprintf("Port number %d saved to global variable\n", port); serverPortNumber = port; tv.tv_sec = 0; // that's right, 0.01 sec. tv.tv_usec = 10000; // I misunderstood what it meant originally. while (dispatcher::loop1(&tv) && !server_terminate) { // basically dumps the content of the result channels through the socket is->manage_results(); if (fileprefs) wp = fileprefs->pollFrequency(); else wp = 100; // reset tv, necessary for Linux tv.tv_sec = 0; // assumption is that wp > 1Hz tv.tv_usec = int(1e6/wp); // controllable amount of time. pthread_testcancel(); } // if we arrive at this point, no sockets are left open if (!server_terminate) { srv_dbgprintf("No more sockets active, shutting down server\n"); pthread_exit((void *)1); } else { srv_dbgprintf("server terminated normally\n"); //pthread_exit((void *)0); } //pthread_debug(); // close server and delete everything is->close(); delete is; is = 0; return (void*)0; } #endif // HAVE_PTHREADS