sfs.cc

Go to the documentation of this file. Or view the newest version in sPHENIX GitHub for file 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);
}