sfs.cc

#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <sys/time.h>
#include <unistd.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <signal.h>
#include <dlfcn.h>
#include <sys/ipc.h>

#include <pthread.h>

#ifdef HAVE_BSTRING_H
#include <bstring.h>
#else
#include <strings.h>
#endif

#include <sys/time.h>

#include <sys/socket.h>
#include <net/if.h>
#include <netdb.h> 

#include <sys/wait.h>
#include <sys/resource.h>


#include <sys/types.h>

#ifdef SunOS
#include <sys/filio.h>
#endif

#include <sys/stat.h>

#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/ioctl.h>

#include <stdio.h>
#include <iostream>
#include <iomanip>

#ifdef HAVE_GETOPT_H
#include <getopt.h>
#endif


typedef unsigned int PHDWORD;
typedef unsigned short SWORD;
typedef unsigned char BYTE;
typedef unsigned int UINT;

#define BUFFERBLOCKSIZE 8192U


#define CTRL_BEGINRUN        1
#define CTRL_ENDRUN          2
#define CTRL_DATA            3
#define CTRL_CLOSE           4
#define CTRL_SENDFILENAME    5
#define CTRL_ROLLOVER        6

#define CTRL_REMOTESUCCESS 100
#define CTRL_REMOTEFAIL    101

#define ROLE_RECEIVED 0
#define ROLE_WRITTEN  1



// initial size 16 Mbytes ( /4 in int units)
#define INITIAL_SIZE (4*1024*1024)


typedef struct 
{
  int dirty;
  int role;
  int bytecount;
  int buffersize;
  PHDWORD *bf;
} bufferstructure;

int run_number = 0;
int verbose = 0;
int sockfd = 0;
int dd_fd = 0;


pthread_mutex_t M_cout;

pthread_mutex_t M_write;
pthread_mutex_t M_done;


pthread_t ThreadId;
pthread_t       tid;
int output_fd = -1;

int the_port = 5001;

int do_not_write = 0;

int RunIsActive = 0; 
int NumberWritten = 0; 
int file_open = 0;


int readn(int , char *, int);
int writen(int , char *, int);

#if defined(SunOS) || defined(Linux) 
void sig_handler(int);
#else
void sig_handler(...);
#endif

void *writebuffers ( void * arg);
int handle_this_child( pid_t pid, const std::string &host);
in_addr_t find_address_from_interface(const char *);

void cleanup(const int exitstatus);


bufferstructure *bf_being_received;
bufferstructure *bf_being_written;
 

using namespace std;


void exitmsg()
{
  cout << "** This is the Super Fast Server :-)." << endl;
  cout << "** usage: sfs " << endl;
  cout << "   -d disable database logging [ db not yet implemented ]" << endl;
  cout << "   -b interface    bind only to this interface" << endl;
  cout << "   -p number       use this port (default 5001)" << endl;
  cout << "   -v increase verbosity" << endl;
  cout << "   -x do not write data to disk (testing)" << endl;
  cout << "  Examples:" << endl;
  cout << "    sfs -b ens801f0      -- listen only on that interface" << endl;
  cout << "    sfs -p 5002          -- listen on port 5002" << endl;
  
  exit(0);
}


//char *s_opcode[] = {
//		    "Invalid code",
//		    "CTRL_BEGINRUN",
//		    "CTRL_ENDRUN",
//		    "CTRL_DATA",
//		    "CTRL_CLOSE",
//		    "CTRL_SENDFILENAME"};

std::string listen_interface;

int main( int argc, char* argv[])
{


#if defined(SunOS) || defined(Linux) 
  struct sockaddr client_addr;
#else
  struct sockaddr_in client_addr;
#endif
  struct sockaddr_in server_addr;

#if defined(SunOS)
  int client_addr_len = sizeof(client_addr);
#else
  unsigned int client_addr_len = sizeof(client_addr);
#endif

  pthread_mutex_init(&M_cout, 0); 
  pthread_mutex_init(&M_write, 0); 
  pthread_mutex_init(&M_done, 0); 

  pthread_mutex_lock(&M_write);

  // by default, we listen on all interfaces
  in_addr_t listen_address = htonl(INADDR_ANY);  

  char c;
  
  while ((c = getopt(argc, argv, "hvdxb:p:")) != EOF)
    {
      switch (c) 
	{

	case 'v':   // verbose
	  verbose++;
	  break;

	case 'h': 
	  exitmsg();
	  break;

	case 'b':   // bind to this interface
	  listen_address = find_address_from_interface(optarg);
	  listen_interface = optarg;
	  break;

	case 'p':   // port number
	  if ( !sscanf(optarg, "%d", &the_port) ) exitmsg();
	  break;

	case 'd':   // no database
	  // databaseflag=1;
	  // cout << "database access enabled" << endl;
	  break;

	case 'x':   // no writing
	  do_not_write = 1;
	  break;


	}
    }



  if (argc > optind)
    {
      sscanf (argv[optind],"%d", &the_port);
    }
  
  // ------------------------
  // now set up the sockets

  if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0 ) 
    {
      cleanup(1);
    }
  //  int xs = 64*1024+21845;
  int xs = 1024*1024;

  int s = setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF,
		     &xs, sizeof(xs));

  if (s)
    {
      perror("Setsockopt:");
    }

  memset( &server_addr, 0, sizeof(server_addr));
  server_addr.sin_family = AF_INET;
  server_addr.sin_addr.s_addr = listen_address;
  server_addr.sin_port = htons(the_port);

  int i;
  
  if ( ( i = bind(sockfd, (struct sockaddr*) &server_addr, sizeof(server_addr))) < 0 )
    {
      perror(" bind ");
      cleanup(1);
    }

  if ( ( i =  listen(sockfd, 100) ) )
    {
      perror(" listen ");
      cleanup(1);
    }

  if (verbose)
    {
      cout << " listening on port " << the_port;
      if (! listen_interface.empty() ) cout << " on interface " << listen_interface;
      cout << endl; 
    }
  
  signal(SIGCHLD, SIG_IGN); 

  pid_t pid;
  struct sockaddr_in out;

  std::string host;
  
  while  (sockfd > 0)
    {
			
      client_addr_len = sizeof(out); 
      dd_fd = accept(sockfd,  (struct sockaddr *) &out, &client_addr_len);
      if ( dd_fd < 0 ) 
	{
	  pthread_mutex_lock(&M_cout);
	  cout  << "error in accept socket" << endl;
	  pthread_mutex_unlock(&M_cout);
	  perror (" accept" );
	  cleanup(1);
	  exit(1);
	}


      char h[512]; 
      getnameinfo((struct sockaddr *) &out, sizeof(struct sockaddr_in), h, 511,
		  NULL, 0, NI_NOFQDN);
      host = h;
      
      if (verbose)
	{
    	  cout << " new connection from " << host << " at " << time(0) << endl;
	}
      
      
      if ( (pid = fork()) == 0 ) 
	{
	  close(sockfd);
	  return handle_this_child( pid, host);
	}
      else
	{
	  close (dd_fd);
	}
    }
}

int handle_this_child( pid_t pid, const std::string &host)
{


  int controlword;
  int local_runnr = 0;

  bufferstructure B0;
  bufferstructure B1;
  
  bf_being_received = &B0;
  bf_being_written = &B1;
  bufferstructure *bf_temp;

  B0.bf = new PHDWORD[INITIAL_SIZE];
  B0.buffersize = INITIAL_SIZE;
  B0.dirty = 0;
  B0.role= ROLE_RECEIVED;
  B0.bytecount = 0;
		      
  B1.bf = new PHDWORD[INITIAL_SIZE];
  B1.buffersize = INITIAL_SIZE;
  B1.dirty = 0;
  B1.role= ROLE_RECEIVED;
  B1.bytecount = 0;
		      
  int i;
  
  // we make a thread that will write out our buffers
  i = pthread_create(&ThreadId, NULL, 
		     writebuffers, 
		     (void *) 0);
  if (i )
    {
      cout << "error in thread create " << i << endl;
      perror("Thread ");
      cleanup(1);
    }
  // else
  //   {
  //     pthread_mutex_lock(&M_cout); 
  //     cout << "write thread created" << endl;
  //     pthread_mutex_unlock(&M_cout);
  //   }
    
  // should be the default, but set it to blocking mode anyway
  i = ioctl (dd_fd,  FIONBIO, 0);
    
  // find out where we were contacted from
    

  int status;
  int xx;

  int go_on = 1;
  while ( go_on)
    {
      if ( (status = readn (dd_fd, (char *) &xx, sizeof(xx)) ) <= 0)
	{
	  cout  << "error in read from socket" << endl;
	  perror ("read " );
	  cleanup(1);
	}
	
      controlword = ntohl(xx);

      // cout << endl;
      // cout << __FILE__ << " " << __LINE__  << " controlword = " << controlword << " ntew: " << xx << " " ;
      // if ( controlword >=0 && controlword <=5) cout  << s_opcode[controlword];
      // cout << endl;
	
      char *p;
      char filename[1024];
      int value, len;
      
      switch (controlword) 
	{

	case  CTRL_SENDFILENAME:
	  // read the length of what we are about to get 
	  readn (dd_fd, (char *) &len, sizeof(int));
	  len  = ntohl(len);
	  // acknowledge... or not
	  //cout  << " filename len  = " << len << endl;
	  if ( len >= 1023)
	    {
	      i = htonl(CTRL_REMOTEFAIL);
	      writen (dd_fd, (char *)&i, sizeof(i));
	      break;
	    }
	  value = readn (dd_fd, filename, len);
	  filename[value] = 0;
	  //cout  << " filename is " << filename << endl;

          i = htonl(CTRL_REMOTESUCCESS);
          writen (dd_fd, (char *)&i, sizeof(i));
	  
	  break;
	  
	case  CTRL_ROLLOVER:

	  // after we receive CTRL_ROLLOVER, we get
	  // - the length of the filename
	  // the actual filename
	  // then we send back the status
	  
	  // read the length of what we are about to get 
	  readn (dd_fd, (char *) &len, sizeof(int));
	  len  = ntohl(len);
	  // acknowledge... or not
	  //      cout << __FILE__ << " " << __LINE__ <<  " filename len  = " << len << endl;
	  if ( len >= 1023)
	    {
	      i = htonl(CTRL_REMOTEFAIL);
	      writen (dd_fd, (char *)&i, sizeof(i));
	      break;
	    }
	  value = readn (dd_fd, filename, len);
	  filename[value] = 0;
	  // cout << __FILE__ << " " << __LINE__  << " filename is " << filename << endl;

	  if (! do_not_write)
	    {
	      close ( output_fd);
	      output_fd =  open(filename,  O_WRONLY | O_CREAT | O_TRUNC | O_EXCL | O_LARGEFILE , 
				S_IRWXU | S_IROTH | S_IRGRP );
	      if (output_fd < 0) 
		{
		  cerr << "file " << filename << " exists, I will not overwrite " << endl;
		  i = htonl(CTRL_REMOTEFAIL);
		  writen (dd_fd, (char *)&i, sizeof(i));
		  break;
		}
	      if (verbose)
		{
		  cout << " opened new rollover file " << filename << endl; 
		}
	    }

	  i = htonl(CTRL_REMOTESUCCESS);
	  writen (dd_fd, (char *)&i, sizeof(i));
	  
	  break;
	  
	case  CTRL_BEGINRUN:
	  
	  readn (dd_fd, (char *) &local_runnr, sizeof(local_runnr) );
	  local_runnr = ntohl(local_runnr);
	  //cout  << " runnumber = " << local_runnr << endl;

	  if (! do_not_write)
	    {
	      output_fd =  open(filename,  O_WRONLY | O_CREAT | O_TRUNC | O_EXCL | O_LARGEFILE , 
				S_IRWXU | S_IROTH | S_IRGRP );
	      if (output_fd < 0) 
		{
		  cerr << "file " << filename << " exists, I will not overwrite " << endl;
		  i = htonl(CTRL_REMOTEFAIL);
		  writen (dd_fd, (char *)&i, sizeof(i));
		  break;
		}
	      if (verbose)
		{
		  cout << " opened new file " << filename << endl; 
		}
	    }

	  bf_being_received = &B0;
	  bf_being_written = &B1;
	    
	  bf_being_received->role = ROLE_RECEIVED;
	  bf_being_received->dirty=0;
	    
	  bf_being_written->role = ROLE_WRITTEN;
	  bf_being_written->dirty = 0;
	    
	  i = htonl(CTRL_REMOTESUCCESS);
	  writen (dd_fd, (char *)&i, sizeof(i));
	  
	  break;
	  





	case  CTRL_DATA:
	  status = readn (dd_fd, (char *) &len, sizeof(i));
	  len = ntohl(len);
	  //cout  << " data! len = " << len << endl;
	  
	  bf_being_received->bytecount = len;
	  if ( (len+sizeof(int)-1)/sizeof(int) > bf_being_received->buffersize)
	    { 
	      delete [] bf_being_received->bf;
	      bf_being_received->buffersize = (len + sizeof(int)-1)/sizeof(int) + 2048;
	      
	      pthread_mutex_lock(&M_cout);
	      //	      cout << "expanding buffer to " << bf_being_received->buffersize << " for host " << host << endl;
	      pthread_mutex_unlock(&M_cout);
	      
	      bf_being_received->bf = new PHDWORD[ bf_being_received->buffersize];
	    }
	  
	  p = (char *) bf_being_received->bf;
	  status = readn (dd_fd, p,   len );
	  if ( len != status)
	    {
	      
	      pthread_mutex_lock(&M_cout);
	      //	      cout << "error on transfer: Expected " << len 
	      //	   << " got status  " << status  << "  for host " << host << endl;
	      perror("ctrl_data");
	      pthread_mutex_unlock(&M_cout);
	      i = htonl(CTRL_REMOTEFAIL);
	    }
	  // store the actual byte length we received
	  bf_being_received->bytecount = len;
	  bf_being_received->dirty = 1;
 
	  //  cout << __FILE__ << " " << __LINE__ << " waiting for write thread " << endl;
	  pthread_mutex_lock(&M_done);
	  // and switch the buffers
	  bf_temp = bf_being_received;
	  bf_being_received = bf_being_written;
	  bf_being_written = bf_temp;
	  bf_being_written->role = ROLE_WRITTEN;
	  bf_being_received->role = ROLE_RECEIVED;
	      
	  //	  cout << __FILE__ << " " << __LINE__ << " switching buffers " << bf_being_received << "  " << bf_being_received << endl;
	  pthread_mutex_unlock(&M_write);
	  
	  i = htonl(CTRL_REMOTESUCCESS);
	  writen (dd_fd, (char *)&i, sizeof(int));
	  
	  break;

	case  CTRL_ENDRUN:
	  //cout  << " endrun signal " << endl;
	  
	  pthread_mutex_lock(&M_done);
	  if (! do_not_write)
	    {
	      close (output_fd);
	      if (verbose)
		{
		  cout << " closed file "  << filename << " at " << time(0) << endl;
		}
	    }

	  i = htonl(CTRL_REMOTESUCCESS);
	  writen (dd_fd, (char *)&i, sizeof(i));
	  pthread_mutex_unlock(&M_done);

	  break;
	  
	case CTRL_CLOSE:
	  close ( dd_fd);

	  // we set go_on to 0 so our loop stops and we return 
	  go_on = 0;
	  if (verbose == 1)
	    {
	      cout << " closed connection from " << host << endl; 
	    }
	  else if (verbose > 1)
	    {
	      cout << " closed connection from "  << host << " at " << time(0) << endl;
	    }
	  break;
	  
	}
		  
    }
  
  if (verbose > 1) cout << " ending thread " << " at " << time(0) << endl;
  
  return 0;
}



void *writebuffers ( void * arg)
{

  while(1)
    {

      pthread_mutex_lock(&M_write);

      if (! do_not_write)
	{
	  pthread_mutex_lock(&M_cout);
	  //cout << __LINE__ << "  " << __FILE__ << " write thread unlocked  " <<  endl;
	  pthread_mutex_unlock(&M_cout);
	  
	  int blockcount = ( bf_being_written->bytecount + BUFFERBLOCKSIZE -1)/BUFFERBLOCKSIZE;
	  int bytecount = blockcount*BUFFERBLOCKSIZE;
	  
	  pthread_mutex_lock(&M_cout);
	  //cout << __LINE__ << "  " << __FILE__ << " write thread unlocked, block count " << blockcount <<  endl;
	  pthread_mutex_unlock(&M_cout);
	  
	  int bytes = writen ( output_fd, (char *) bf_being_written->bf , bytecount );
	  if ( bytes != bytecount)
	    {
	      pthread_mutex_lock(&M_cout);
	      cout << __LINE__ << "  " << __FILE__ << " write error " << bytes << "  " << bytecount <<  endl;
	      pthread_mutex_unlock(&M_cout);
	      bf_being_written->dirty = -1;  // mark as "error"
	    }
	  else
	    {
	      //      usleep(1000000);
	      bf_being_written->dirty = 0;
	    }
	}
      else
	{ 
	  bf_being_written->dirty = 0;
	}
      pthread_mutex_unlock(&M_done);
      
    }
  return 0;
}




int readn (int fd, char *ptr, int nbytes)
{

  int nread, nleft;
  //int nleft, nread;
  nleft = nbytes;
  while ( nleft>0 )
    {
      nread = read (fd, ptr, nleft);
      if ( nread <= 0 ) 
	{
	  return nread;
	}

#ifdef FRAGMENTMONITORING
      history[hpos++] = nread;
#endif
      nleft -= nread;
      ptr += nread;
    }

#ifdef FRAGMENTMONITORING
  if ( hpos >1 ) 
    {
      cout << "Fragmented transfer of " << nbytes << "bytes: ";
      for ( int i=0; i<hpos; i++)
	{
	  cout << " " << history[i]<< ",";
	}
      cout << endl;
    }
#endif
  return (nbytes-nleft);
}
  
int writen (int fd, char *ptr, int nbytes)
{
  int nleft, nwritten;
  nleft = nbytes;
  while ( nleft>0 )
    {
      nwritten = write (fd, ptr, nleft);
      if ( nwritten < 0 ) 
	return nwritten;

      nleft -= nwritten;
      ptr += nwritten;
    }
  return (nbytes-nleft);
}


in_addr_t find_address_from_interface(const char *interface)
{
  int fd;
  struct ifreq ifr;

  // check the no one plays tricks with us...
  if ( strlen(interface) >= IFNAMSIZ)
    {
      cerr << " Interface name too long, ignoring "<< endl;
      return htonl(INADDR_ANY);
    }
  
  fd = socket(AF_INET, SOCK_DGRAM, 0);

  // I want to find an IPv4 IP address
  ifr.ifr_addr.sa_family = AF_INET;

  // I want IP address attached to "interface"
  strncpy(ifr.ifr_name, interface, IFNAMSIZ-1);

  ioctl(fd, SIOCGIFADDR, &ifr);

  close(fd);
  in_addr_t a =  ((struct sockaddr_in *)&ifr.ifr_addr)->sin_addr.s_addr;

  if (verbose)  cout << "  binding to address " << inet_ntoa(  ((struct sockaddr_in *)&ifr.ifr_addr)->sin_addr ) << endl;
  return a;
}



void cleanup( const int exitstatus)
{
  close (dd_fd);

  exit(exitstatus);
}