src/net/lognet.cpp

   1 #include "genint1.h"
   2 #include "comm.h"
   3
   4 #include <sys/socket.h>
   5 #include <sys/un.h>
   6 #include <netinet/in.h>
   7 #include <errno.h>
   8 #include <netdb.h>
   9 #include <arpa/inet.h>
  10
  11 #include <sys/types.h>
  12 #include <sys/time.h>
  13 #include <stdlib.h>
  14 #include <stdio.h>
  15 #include <fcntl.h>
  16 #include <signal.h>
  17 #include <sys/stat.h>
  18 #include <string.h>
  19 #include <qlist.h>
  20 #include <qfile.h>
  21 #include <qstring.h>
  22 #include <qstringlist.h>
  23 #include <unistd.h>
  24
  25 #include <libconfig.h>
  26
  27 #define REMOTE_PATH "REMOTE"
  28 #define MAXLINKS 30
  29 #define LOGPORT 3600
  30 #define CODEPORT 3700
  31 #define CODEPORT1 3800
  32 #define MAXINTERP 10
  33 #define FILE_BUFFER_SIZE 2048
  34
  35
  36
  37 // ************** Interpreter slot *******************
  38 class INTlink
  39 {
  40  public:
  41   int sock,ID;
  42   bool connected;
  43   INTlink();
  44
  45 };
  46
  47 INTlink::INTlink()
  48 {
  49  connected=FALSE;
  50  sock=0;
  51 }
  52
  53
  54 // ********************  Network slot ********************
  55 class NETlink
  56 {
  57  public:
  58  int sock;
  59  bool connected,code_transmit;
  60  char addr[255];
  61
  62  int node_number;
  63  int aliases[5];
  64
  65  FILE *CodeFile;
  66  char CodeName[255];
  67  long CodeSize;
  68
  69
  70  NETlink();
  71 };
  72
  73 NETlink::NETlink()
  74 {
  75  int i;
  76  for(i=0;i<5;i++) aliases[i]=-1;
  77  connected=FALSE;
  78  sock=0;
  79  code_transmit=FALSE;
  80 }
  81
  82
  83
  84
  85 //********************** NET Module ****************************
  86
  87 class NETMOD
  88 {
  89 public:
  90  int kernel_sock,listen_sock;
  91  bool all_connected,local_mode;
  92  int to_connect,MyNode;
  93  char kername[256];
  94
  95
  96  QList<INTlink> Interpreters; // List of the Interpeter slots
  97  QList<NETlink> Links;       //  List of the Network slots
  98
  99  NETMOD(char*);
 100
 101  void load_config(char*);
 102  void write_at_console(char*);
 103  void send_to_kernel(MESSAGE*);
 104  void sock_reopen(NETlink*);
 105  void send_connect_info(NETlink*);
 106  void send_accept_info(NETlink*);
 107  void send_to_node(NETlink*,MESSAGE*);
 108  void send_to_int(MESSAGE*);
 109  void send_code_ack(NETlink*);
 110  void send_to_all(MESSAGE *);
 111
 112  NETlink *findNETlink(int node);
 113  INTlink *findINTlink(int id);
 114  void transmit_file(int ton, char *fname, int fromINT);
 115  void propagate_msg(MESSAGE *msg);
 116  void check_node(int,int);
 117
 118  void run();
 119  void exit_sequence();
 120  void disconnect_seq();
 121  void connect_seq(char*);
 122  void accept_connection();
 123  void get_internal();
 124  void remote_messages();
 125  void check_links();
 126  void get_message(NETlink*);
 127  void conn_info(int);
 128
 129  void doitall(); // 2010
 130 };
 131
 132
 133 NETMOD::NETMOD(char *kernel_name)
 134 {
 135  int i,len,on;
 136  struct sockaddr_in svr;
 137  struct sockaddr_un svr1;
 138  MESSAGE m;
 139  char s[256];
 140
 141
 142  Links.clear();
 143  Interpreters.clear();
 144
 145  bzero(&svr, sizeof(svr));
 146  listen_sock = socket(AF_INET, SOCK_STREAM, 0);
 147  svr.sin_family = AF_INET;
 148  svr.sin_addr.s_addr = INADDR_ANY;
 149  svr.sin_port = htons(LOGPORT);
 150  bind(listen_sock, (struct sockaddr*)&svr, sizeof(svr));
 151  listen(listen_sock,5);
 152  fcntl(listen_sock, F_SETFL,O_NONBLOCK | fcntl(listen_sock, F_GETFL,0));
 153
 154  to_connect=0;
 155  all_connected=FALSE;
 156  load_config("vlp.cfg");
 157
 158  kernel_sock = socket(AF_UNIX,SOCK_STREAM,0);
 159  bzero(&svr1,sizeof(svr1));
 160  svr1.sun_family = AF_UNIX;
 161  strcpy(svr1.sun_path,kernel_name);
 162  strcpy(kername,kernel_name);
 163  len = strlen(svr1.sun_path)+sizeof(svr1.sun_family);
 164  i = connect(kernel_sock,(struct sockaddr*)&svr1,len);
 165  if (i==0)
 166  fcntl(kernel_sock,F_SETFL, O_NONBLOCK|fcntl(kernel_sock,F_GETFL,0));
 167  on=1;
 168  setsockopt(kernel_sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 169  m.msg_type = MSG_NET;
 170  m.param.pword[0] = NET_NODE;
 171  m.param.pword[1] = MyNode;
 172  send_to_kernel(&m);
 173
 174  // if (regme) regme_sequence();
 175
 176  if (to_connect > 0){
 177  write_at_console("Connecting remote VLPs...");
 178  while (!all_connected) check_links();
 179                     }
 180  sprintf(s,"Local node number %d",MyNode);
 181  write_at_console(s);
 182 }
 183
 184 // #####################  Load configuration ##########################
 185
 186 void NETMOD::load_config(char *fname)
 187 {
 188   config_t cfg;
 189   config_setting_t *setting;
 190   int on,k=0;
 191   NETlink *pomlink;
 192
 193   config_init(&cfg);
 194
 195
 196   /* Hack for checking if file exists without using external libs.*/
 197   FILE * file = fopen(fname, "rt");
 198   if (!file) {
 199     fprintf(stderr, "Error: Cannot load configuration file %s!\n", fname);
 200     write_at_console("Cannot load configuration file!");
 201     fclose(file);
 202     exit(3);
 203   }
 204
 205   /* Read the file. If there is an error, report it and exit. */
 206   if(!config_read(&cfg, file))
 207   {
 208     fprintf(stderr, "%s: In file %s, line %d\n",
 209         config_error_text(&cfg),
 210         config_error_file(&cfg),
 211         config_error_line(&cfg));
 212     config_destroy(&cfg);
 213     fclose(file);
 214     exit(3);/* from original code. */
 215   }
 216
 217   setting = config_lookup(&cfg, "node_number");
 218   if(setting)
 219   {
 220     MyNode = config_setting_get_int(setting);
 221   }
 222   /* else */
 223   if (!setting || MyNode==-1)
 224   {
 225     fprintf(stderr, "%s! In file %s, '%s' was not found.\n",
 226       "Error",
 227       fname,
 228       "node_number");
 229     write_at_console("Node number must be specified");
 230     config_destroy(&cfg);
 231     fclose(file);
 232     exit(1);
 233   }
 234
 235   setting = config_lookup(&cfg, "host");
 236   if(setting) {
 237     k++;
 238     pomlink = new NETlink;
 239
 240     switch(config_setting_type(setting)) {
 241       case CONFIG_TYPE_STRING:/* TODO: Deprecated. Made for back compatibility. */
 242         strncpy(pomlink->addr, config_setting_get_string(setting), 255);
 243         break;
 244       case CONFIG_TYPE_ARRAY:
 245         strncpy(pomlink->addr, config_setting_get_string_elem(setting, 0), 255);
 246         break;
 247       default:
 248         fprintf(stderr, "%s! In file %s, bad entry type for %s. Will not be read.\n"
 249           "Fatal error",
 250           fname,
 251           "host");
 252         config_destroy(&cfg);
 253         fclose(file);
 254         exit(1);
 255     }
 256     pomlink->connected = FALSE;
 257     pomlink->sock = socket(AF_INET, SOCK_STREAM, 0);
 258     fcntl(pomlink->sock, F_SETFL,O_NONBLOCK | fcntl(pomlink->sock,F_GETFL,0));
 259     on=1;
 260     setsockopt(pomlink->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 261     Links.append(pomlink);
 262     to_connect++;
 263   }
 264   else {
 265     fprintf(stderr, "%s! In file %s, '%s' was not found.\n",
 266         "Warning",
 267         fname,
 268         "host");
 269   }
 270
 271   config_destroy(&cfg);
 272   fclose(file);
 273
 274   if (k==0) all_connected=TRUE;
 275 }
 276
 277
 278
 279 void NETMOD::write_at_console(char *s)
 280 {
 281  MESSAGE msg;
 282
 283  msg.msg_type = MSG_NET;
 284  msg.param.pword[0] = NET_CSWRITELN;
 285  strcpy(msg.param.pstr,s);
 286  send_to_kernel(&msg);
 287 }
 288
 289 void NETMOD::send_to_kernel(MESSAGE *msg)
 290 {
 291  write(kernel_sock,msg,sizeof(MESSAGE));
 292 }
 293
 294 void NETMOD::send_to_node(NETlink *lnk, MESSAGE *msg)
 295 {
 296  msg->msg_type = MSG_NET;
 297 // msg2netmsg(msg);
 298 if (lnk->sock)
 299  write(lnk->sock,msg,sizeof(MESSAGE));
 300 }
 301
 302 void NETMOD::send_to_int(MESSAGE *msg)
 303 {
 304  INTlink *pomlink;
 305
 306  pomlink = findINTlink(msg->param.pword[5]);
 307  if (pomlink!=NULL) write(pomlink->sock,msg,sizeof(MESSAGE));
 308 }
 309
 310
 311
 312 void NETMOD::accept_connection()
 313 {
 314  unsigned int sz;
 315  int nsock, on;
 316  struct sockaddr_in svr;
 317  fd_set rset,wset;
 318  struct timeval tout = {0,0};
 319  NETlink *pomlink;
 320
 321  FD_ZERO(&rset);FD_ZERO(&wset);
 322  FD_SET(listen_sock,&rset);
 323
 324  if (select(listen_sock+1,&rset,&wset,0,(struct timeval *)&tout)>0)
 325   if (FD_ISSET(listen_sock,&rset))
 326  {
 327
 328 /* accept connection on listen socket */
 329  sz = sizeof(svr);
 330  bzero(&svr, sizeof(svr));
 331  nsock = accept(listen_sock, (struct sockaddr*)&svr, &sz);
 332
 333  if (nsock>0)
 334   {
 335
 336    /* i<0 someone wants to connect us */
 337
 338        pomlink = new NETlink;
 339        strcpy(pomlink->addr,inet_ntoa(svr.sin_addr));
 340        pomlink->sock = nsock;
 341        pomlink->connected = TRUE;
 342        fcntl(pomlink->sock, F_SETFL,O_NONBLOCK | fcntl(pomlink->sock,
 343            F_GETFL,0));
 344        on=1;
 345        setsockopt(pomlink->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 346        Links.append(pomlink);
 347   } /* nsock > 0 */
 348 } /* ISSET */
 349 }
 350
 351
 352 void NETMOD::check_node(int n, int sc)
 353 {
 354  MESSAGE m;
 355  NETlink *pomlink;
 356
 357  m.msg_type = MSG_NET;
 358  m.param.pword[0] = NET_NODE_EXIST;
 359
 360  pomlink = Links.first();
 361  m.param.pword[1] = 0;
 362  while (pomlink!=NULL)
 363  {
 364   if ( pomlink->node_number==n )
 365    {
 366      m.param.pword[1] = 1;
 367      strcpy(m.param.pstr,pomlink->addr);
 368      break;
 369    }
 370   pomlink = Links.next();
 371  }
 372  write(sc,&m,sizeof(MESSAGE));
 373 }
 374
 375
 376 // ************* Internal message from kernel or INT *******************
 377
 378 void NETMOD::get_internal()
 379 {
 380  int nr,nrset;
 381  MESSAGE msg;
 382  int si, sj;
 383  fd_set readset,writeset;
 384  struct timeval tout={0,0};
 385  INTlink *pomlink;
 386  struct sockaddr_un svr;
 387
 388
 389
 390  FD_ZERO(&readset);FD_ZERO(&writeset);
 391  FD_SET(kernel_sock,&readset);
 392  nrset=kernel_sock;
 393
 394  pomlink = Interpreters.first();
 395  while (pomlink!=NULL)
 396  {
 397   FD_SET(pomlink->sock,&readset);
 398   if (nrset<pomlink->sock) nrset=pomlink->sock;
 399   pomlink=Interpreters.next();
 400  }
 401
 402  if (select(nrset+1,&readset,&writeset,0,(struct timeval *)&tout)>0)
 403  {
 404
 405 /* Check request sockets */
 406  pomlink = Interpreters.first();
 407  while (pomlink!=NULL)
 408  {
 409    if (FD_ISSET(pomlink->sock,&readset))
 410    {
 411      nr = read(pomlink->sock,&msg,sizeof(MESSAGE));
 412      if (nr>0)
 413       {
 414           if (msg.msg_type == MSG_NET)
 415              switch(msg.param.pword[0])
 416                {
 417                  case NET_PROPAGATE:propagate_msg(&msg);break;
 418                  case NET_NODE_EXIST:check_node(msg.param.pword[1],pomlink->sock);break;
 419                  case NET_GET_INFO: conn_info(pomlink->sock);break;
 420                  case NET_NODES_NUM: msg.param.pword[0]=NET_NODES_NUM_RESPONSE;
 421                                      msg.param.pword[1]=Links.count();
 422                                      write(pomlink->sock,&msg,sizeof(MESSAGE));
 423                                      break;
 424                }/* switch */
 425       }
 426    } /* ISSET */
 427    pomlink=Interpreters.next();
 428  } // while
 429
 430 /* Check internal socket */
 431  if (FD_ISSET(kernel_sock,&readset))
 432  {
 433   nr = read(kernel_sock, &msg, sizeof(MESSAGE));
 434   if (nr>0)
 435   {
 436    if (msg.msg_type == MSG_NET)
 437    {
 438      switch(msg.param.pword[0])
 439      {
 440        case NET_TRANSMIT_CODE:
 441                  transmit_file(msg.param.pword[2],msg.param.pstr,msg.param.pword[1]);
 442                  break;
 443        case NET_EXIT: { disconnect_seq();exit_sequence();}
 444                    break;
 445        case NET_GET_INFO: conn_info(kernel_sock);break;
 446        case NET_PROPAGATE:propagate_msg(&msg);break;
 447        case NET_DISCONNECT:disconnect_seq();
 448                    break;
 449        case NET_NODE_EXIST: check_node(msg.param.pword[1],kernel_sock);break;
 450        case NET_CONNECT_TO: connect_seq(msg.param.pstr);
 451
 452         } /* end switch */
 453     } /* MSg_NET */
 454
 455    if (msg.msg_type == MSG_VLP)
 456     switch(msg.param.pword[0])
 457     {
 458      case VLP_REGINT:
 459         {
 460            pomlink = new INTlink;
 461            pomlink->sock = socket(AF_UNIX,SOCK_STREAM,0);
 462            bzero(&svr,sizeof(svr));
 463            svr.sun_family = AF_UNIX;
 464            strcpy(svr.sun_path,msg.param.pstr);
 465            si = strlen(svr.sun_path)+sizeof(svr.sun_family);
 466            sj = connect(pomlink->sock,(struct sockaddr*)&svr,si);
 467            if (sj==0)
 468             fcntl(pomlink->sock,F_SETFL, O_NONBLOCK|
 469                    fcntl(pomlink->sock,F_GETFL,0));
 470            int on=1;
 471            setsockopt(pomlink->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 472            pomlink->ID = msg.param.pword[1];
 473            pomlink->connected=TRUE;
 474            Interpreters.append(pomlink);
 475
 476          };break;
 477      case VLP_INTERPRETER_DOWN:
 478         {
 479           pomlink = findINTlink(msg.param.pword[1]);
 480           if (pomlink!=NULL)
 481           {
 482            close(pomlink->sock);
 483            Interpreters.remove(pomlink);
 484           }
 485          };break;
 486
 487      break;
 488     } /* MSg_VLP */
 489    }
 490   }/* ISSET */
 491  } /* select >0 */
 492
 493
 494 }
 495
 496 void NETMOD::get_message(NETlink *lnk)
 497 {
 498  int nr;
 499  MESSAGE msg;
 500  char pomstr[80];
 501  int rdbt,rd,sz,j,psock;
 502  struct sockaddr_in svr;
 503  unsigned char buffer[FILE_BUFFER_SIZE];
 504  protdescr proto;
 505
 506 if (lnk->connected)
 507 {
 508   nr = read(lnk->sock,&msg, sizeof(MESSAGE));
 509   if (nr>0)
 510    {
 511 //     netmsg2msg(&msg);
 512      if (msg.msg_type == MSG_NET)
 513       {
 514         switch(msg.param.pword[0])
 515         {
 516          case NET_CCD_START:
 517
 518                      lnk->code_transmit = TRUE;
 519                      sprintf(pomstr,"%s/%s",REMOTE_PATH,msg.param.pstr);
 520                      strcat(pomstr,".ccd");
 521                      lnk->CodeFile = fopen(pomstr,"wb");
 522                      if ( lnk->CodeFile == NULL) { write_at_console("Cannot open file\n");
 523                      lnk->code_transmit=FALSE;}
 524                      lnk->CodeSize=msg.param.pword[1];
 525                      psock = socket(AF_INET, SOCK_STREAM, 0);
 526                      bzero(&svr, sizeof(svr));
 527                      svr.sin_family = AF_INET;
 528                      svr.sin_port = htons(CODEPORT);
 529                      svr.sin_addr.s_addr = inet_addr(lnk->addr);
 530                      j = connect(psock, (struct sockaddr*)&svr, sizeof(svr));
 531                      if ( j==0)
 532                      {
 533                       //fcntl(psock, F_SETFL,O_NONBLOCK | fcntl(psock, F_GETFL,0));
 534                        sz=0;
 535                        while (sz<lnk->CodeSize)
 536                        {
 537                          rd = read(psock,&buffer,sizeof(buffer));
 538                          rdbt = fwrite(&buffer,sizeof(unsigned char),rd,lnk->CodeFile);
 539                          sz=sz+rd;
 540                         }
 541                        close(psock);
 542                        fclose(lnk->CodeFile);
 543                       }
 544
 545                      break;
 546
 547          case NET_PCD_START:
 548
 549                      lnk->code_transmit = TRUE;
 550                      sprintf(pomstr,"%s/%s",REMOTE_PATH,msg.param.pstr);
 551                      strcat(pomstr,".pcd");
 552                      lnk->CodeFile = fopen(pomstr,"wb");
 553                      if ( lnk->CodeFile == NULL) { write_at_console("Cannot open file\n");
 554                      lnk->code_transmit=FALSE;}
 555                      lnk->CodeSize=msg.param.pword[1];
 556                      psock = socket(AF_INET, SOCK_STREAM, 0);
 557                      bzero(&svr, sizeof(svr));
 558                      svr.sin_family = AF_INET;
 559                      svr.sin_port = htons(CODEPORT1);
 560                      svr.sin_addr.s_addr = inet_addr(lnk->addr);
 561                      j = connect(psock, (struct sockaddr*)&svr, sizeof(svr));
 562                      if ( j==0)
 563                      {
 564                       //fcntl(psock, F_SETFL,O_NONBLOCK | fcntl(psock, F_GETFL,0));
 565                        sz=0;
 566                        while (sz<lnk->CodeSize)
 567                        {
 568                          rd = read(psock,&proto,sizeof(proto));
 569                          rdbt = fwrite(&proto,sizeof(unsigned char),rd,lnk->CodeFile);
 570                          sz=sz+rd;
 571                         }
 572                        close(psock);
 573                        fclose(lnk->CodeFile);
 574                       }
 575
 576                      break;
 577
 578          case NET_CONNECT:
 579
 580                      sprintf(pomstr,"Node: %d Addr: %s",msg.param.pword[1],
 581                      lnk->addr);
 582                      lnk->node_number = msg.param.pword[1];
 583                      write_at_console(pomstr);
 584                      send_accept_info(lnk);
 585                      break;
 586          case NET_ACCEPT:
 587
 588                     sprintf(pomstr,"Node: %d Addr: %s",msg.param.pword[1],
 589                     lnk->addr);
 590                     lnk->node_number = msg.param.pword[1];
 591                     write_at_console(pomstr);
 592                     break;
 593          case NET_DISCONNECT:
 594
 595                     sprintf(pomstr,"Node: %d disconnected",msg.param.pword[1]);
 596                     write_at_console(pomstr);
 597                     ::close(lnk->sock);
 598                     Links.remove(lnk);
 599                     delete(lnk);
 600                     break;
 601          case NET_PROPAGATE:
 602
 603                if (msg.param.pword[1] == MSG_VLP) send_to_kernel(&msg);
 604                else if (msg.param.pword[1] == MSG_INT) send_to_int(&msg);
 605                break;
 606
 607                 } /* end switch */
 608
 609        }
 610      } /* nr > 0 */
 611
 612 } /* end if used */
 613 }
 614
 615
 616
 617 void NETMOD::remote_messages()
 618 {
 619  int max;
 620  fd_set rset,wset;
 621  struct timeval tout={0,0};
 622  NETlink *pomlink;
 623
 624  FD_ZERO(&rset);FD_ZERO(&wset);
 625  max=0;
 626
 627  pomlink = Links.first();
 628  while (pomlink!=NULL)
 629  {
 630   if (pomlink->connected)
 631   {
 632     FD_SET(pomlink->sock,&rset);
 633     if  (max<pomlink->sock) max=pomlink->sock;
 634   }
 635   pomlink=Links.next();
 636  }
 637
 638  if (select(max+1,&rset,&wset,0,(struct timeval *)&tout)>0)
 639  {
 640   pomlink=Links.first();
 641   while (pomlink!=NULL)
 642   {
 643    if (FD_ISSET(pomlink->sock,&rset)) get_message(pomlink);
 644    pomlink=Links.next();
 645   }
 646  }
 647
 648 }
 649
 650
 651 /******************************    2010 ********************************************/
 652 void NETMOD::doitall() {
 653  unsigned int sz;
 654  int nsock,max=0,on, nr, si, sj;
 655  struct sockaddr_in svr;
 656  fd_set rset;
 657  NETlink *pomlink;
 658  INTlink *pomlink2;
 659  struct sockaddr_un svr2;
 660  MESSAGE msg;
 661
 662
 663  FD_ZERO(&rset);
 664  FD_SET(listen_sock,&rset);
 665  max = listen_sock;
 666  FD_SET(kernel_sock,&rset);
 667  if (max<kernel_sock) { max=kernel_sock;}
 668  pomlink2 = Interpreters.first();
 669  while (pomlink2!=NULL)
 670  {
 671   FD_SET(pomlink2->sock,&rset);
 672   if (max<pomlink2->sock) max=pomlink2->sock;
 673   pomlink2=Interpreters.next();
 674  }
 675  pomlink = Links.first();
 676  while (pomlink!=NULL)
 677  {
 678   if (pomlink->connected)
 679   {
 680     FD_SET(pomlink->sock,&rset);
 681     if  (max<pomlink->sock) max=pomlink->sock;
 682   }
 683   pomlink=Links.next();
 684  }
 685
 686  // odczyt
 687  if (select(max+1,&rset,0,0,NULL) > 0) {
 688     // accept connection
 689        if (FD_ISSET(listen_sock,&rset))
 690         {
 691           /* accept connection on listen socket */
 692           sz = sizeof(svr);
 693           bzero(&svr, sizeof(svr));
 694           nsock = accept(listen_sock, (struct sockaddr*)&svr, &sz);
 695           if (nsock>0)
 696           {
 697             pomlink = new NETlink;
 698             strcpy(pomlink->addr,inet_ntoa(svr.sin_addr));
 699             pomlink->sock = nsock;
 700             pomlink->connected = TRUE;
 701             fcntl(pomlink->sock, F_SETFL,O_NONBLOCK | fcntl(pomlink->sock,F_GETFL,0));
 702             on=1;
 703             setsockopt(pomlink->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 704             Links.append(pomlink);
 705            } /* nsock > 0 */
 706         } /* ISSET */
 707
 708     // get internal message
 709        /* Check request sockets */
 710        pomlink2 = Interpreters.first();
 711        while (pomlink2!=NULL)
 712        {
 713          if (FD_ISSET(pomlink2->sock,&rset))
 714          {
 715            nr = read(pomlink2->sock,&msg,sizeof(MESSAGE));
 716            if (nr>0)
 717            {
 718              if (msg.msg_type == MSG_NET)
 719                switch(msg.param.pword[0])
 720                {
 721                  case NET_PROPAGATE:propagate_msg(&msg); break;
 722                  case NET_NODE_EXIST:check_node(msg.param.pword[1],pomlink2->sock);break;
 723                  case NET_GET_INFO: conn_info(pomlink2->sock);break;
 724                  case NET_NODES_NUM: msg.param.pword[0]=NET_NODES_NUM_RESPONSE;
 725                                      msg.param.pword[1]=Links.count();
 726                                      write(pomlink2->sock,&msg,sizeof(MESSAGE));
 727                                      break;
 728                }/* switch */
 729            }
 730          } /* ISSET */
 731          pomlink2=Interpreters.next();
 732        } // while
 733        /* Check internal socket */
 734        if (FD_ISSET(kernel_sock,&rset))
 735        {
 736           nr = read(kernel_sock, &msg, sizeof(MESSAGE));
 737           if (nr>0)
 738           {
 739             if (msg.msg_type == MSG_NET)
 740             {
 741               switch(msg.param.pword[0])
 742               {
 743                  case NET_TRANSMIT_CODE:
 744
 745                  transmit_file(msg.param.pword[2],msg.param.pstr,msg.param.pword[1]);
 746
 747                  break;
 748                  case NET_EXIT: {  disconnect_seq();exit_sequence();}
 749                  break;
 750                  case NET_GET_INFO: conn_info(kernel_sock);break;
 751                  case NET_PROPAGATE:propagate_msg(&msg);break;
 752                  case NET_DISCONNECT:disconnect_seq();
 753                    break;
 754                  case NET_NODE_EXIST: check_node(msg.param.pword[1],kernel_sock);break;
 755                  case NET_CONNECT_TO: connect_seq(msg.param.pstr);
 756               } /* end switch */
 757             } /* MSG_NET */
 758             if (msg.msg_type == MSG_VLP)
 759               switch(msg.param.pword[0])
 760               {
 761                  case VLP_REGINT:
 762                  {
 763
 764                     pomlink2 = new INTlink;
 765                     pomlink2->sock = socket(AF_UNIX,SOCK_STREAM,0);
 766                     bzero(&svr2,sizeof(svr2));
 767                     svr2.sun_family = AF_UNIX;
 768                     strcpy(svr2.sun_path,msg.param.pstr);
 769                     si = strlen(svr2.sun_path)+sizeof(svr2.sun_family);
 770                     sj = connect(pomlink2->sock,(struct sockaddr*)&svr2,si);
 771                     if (sj==0)
 772                       fcntl(pomlink2->sock,F_SETFL, O_NONBLOCK|fcntl(pomlink2->sock,F_GETFL,0));
 773                     int on=1;
 774                     setsockopt(pomlink2->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 775                     pomlink2->ID = msg.param.pword[1];
 776                     pomlink2->connected=TRUE;
 777                     Interpreters.append(pomlink2);
 778
 779                  };break;
 780                  case VLP_INTERPRETER_DOWN:
 781                  {
 782
 783                    pomlink2 = findINTlink(msg.param.pword[1]);
 784                    if (pomlink2!=NULL)
 785                    {
 786                      close(pomlink2->sock);
 787                      Interpreters.remove(pomlink2);
 788                    }
 789                  };break;
 790               } /* MSg_VLP */
 791            } // nr > 0
 792        }/* ISSET */
 793
 794     // get remote message
 795
 796     pomlink=Links.first();
 797     while (pomlink!=NULL)
 798     {
 799       if (FD_ISSET(pomlink->sock,&rset)) get_message(pomlink);
 800       pomlink=Links.next();
 801     }
 802  } // select
 803
 804
 805 }
 806 /******************************    END 2010 ********************************************/
 807
 808
 809
 810 void NETMOD::propagate_msg(MESSAGE *msg)
 811 {
 812  char ss[255];
 813  NETlink *pomlink;
 814
 815  pomlink = findNETlink(msg->param.pword[4]);
 816  if ((pomlink!=NULL)&&(pomlink->connected))
 817   send_to_node(pomlink,msg);
 818  else {
 819        if (msg->param.pword[1]==MSG_INT)
 820        {
 821         send_to_int(msg);
 822         }
 823        else
 824        {
 825        sprintf(ss,"Not connected to Node %d",msg->param.pword[4]);
 826        write_at_console(ss);
 827        }
 828       }
 829
 830 }
 831
 832
 833 void NETMOD::connect_seq(char *a)
 834 {
 835  NETlink *pom;
 836  struct sockaddr_in svr;
 837  int j,on;
 838
 839
 840  pom=Links.first();
 841  while (pom!=NULL)
 842  {
 843   if (strcmp(pom->addr,a)==0) return;
 844   pom=Links.next();
 845   }
 846      pom = new NETlink;
 847      strcpy(pom->addr,a);
 848      pom->connected = FALSE;
 849      pom->sock = socket(AF_INET, SOCK_STREAM, 0);
 850      bzero(&svr, sizeof(svr));
 851      svr.sin_family = AF_INET;
 852      svr.sin_port = htons(LOGPORT);
 853      svr.sin_addr.s_addr = inet_addr(pom->addr);
 854      j = connect(pom->sock, (struct sockaddr*)&svr, sizeof(svr));
 855      if ( j==0)
 856           { pom->connected = TRUE;
 857            fcntl(pom->sock, F_SETFL,O_NONBLOCK | fcntl(pom->sock,
 858            F_GETFL,0));
 859            on=1;
 860            setsockopt(pom->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 861            send_connect_info(pom);
 862            Links.append(pom);
 863           }
 864       else write_at_console("Connection failed");
 865 }
 866
 867 void NETMOD::check_links()
 868 {
 869
 870 int j=1;
 871 struct sockaddr_in svr;
 872 NETlink *pomlink;
 873
 874 /* connect to all other nodes */
 875 if (!all_connected)
 876 {
 877  pomlink=Links.first();
 878  while (pomlink!=NULL)
 879  {
 880  if ( !(pomlink->connected) )
 881  {
 882    bzero(&svr, sizeof(svr));
 883    svr.sin_family = AF_INET;
 884    svr.sin_port = htons(LOGPORT);
 885    svr.sin_addr.s_addr = inet_addr(pomlink->addr);
 886
 887    j = connect(pomlink->sock, (struct sockaddr*)&svr, sizeof(svr));
 888    if ( j==0)
 889           { pomlink->connected = TRUE;
 890            fcntl(pomlink->sock, F_SETFL,O_NONBLOCK | fcntl(pomlink->sock,
 891            F_GETFL,0));
 892            send_connect_info(pomlink);
 893
 894            }
 895    if (errno == ECONNREFUSED)
 896            sock_reopen(pomlink);
 897  } // not connected
 898   pomlink=Links.next();
 899  } //while
 900
 901 }//if
 902 all_connected=TRUE;
 903 pomlink=Links.first();
 904 while(pomlink!=NULL)
 905 {
 906  if (pomlink->connected==FALSE) {all_connected=FALSE;break;}
 907  pomlink=Links.next();
 908 }
 909
 910 }
 911
 912 void NETMOD::sock_reopen(NETlink *lnk)
 913 {
 914  int on=1;
 915
 916  close(lnk->sock);
 917  lnk->sock = socket(AF_INET, SOCK_STREAM, 0);
 918  fcntl(lnk->sock, F_SETFL,O_NONBLOCK | fcntl(lnk->sock,
 919            F_GETFL,0));
 920  setsockopt(lnk->sock,IPPROTO_TCP,TCP_NODELAY,(char*)&on,sizeof(on));
 921 }
 922
 923
 924 // **************** Acknowledges *************************
 925
 926 void NETMOD::send_connect_info(NETlink *lnk)
 927 {
 928  MESSAGE m;
 929
 930  m.param.pword[0] = NET_CONNECT;
 931  m.param.pword[1] = MyNode;
 932  m.msg_type = MSG_NET;
 933 // msg2netmsg(&m);
 934  if (lnk->sock)
 935  write(lnk->sock,&m,sizeof(MESSAGE));
 936 }
 937
 938 void NETMOD::send_accept_info(NETlink *lnk)
 939 {
 940  MESSAGE m;
 941
 942  m.param.pword[0] = NET_ACCEPT;
 943  m.param.pword[1] = MyNode;
 944  m.msg_type = MSG_NET;
 945 // msg2netmsg(&m);
 946  if (lnk->sock)
 947  write(lnk->sock,&m,sizeof(MESSAGE));
 948 }
 949
 950 void NETMOD::send_code_ack(NETlink *lnk)
 951 {
 952 MESSAGE m;
 953
 954  m.param.pword[0] = NET_CODESTREAM_OK;
 955  m.msg_type = MSG_NET;
 956 // msg2netmsg(&m);
 957  if (lnk->sock)
 958  write(lnk->sock,&m,sizeof(MESSAGE));
 959 }
 960
 961
 962 void NETMOD::send_to_all(MESSAGE *msg)
 963 {
 964  NETlink *pomlink;
 965  pomlink=Links.first();
 966  while (pomlink!=NULL)
 967  {
 968   write(pomlink->sock,msg,sizeof(MESSAGE));
 969   pomlink=Links.next();
 970  }
 971 }
 972
 973
 974
 975
 976 void NETMOD::run()
 977 {
 978   while(1)
 979  {
 980   /*accept_connection();
 981   get_internal();
 982   remote_messages(); */
 983   // 2010
 984   doitall();
 985  }
 986 }
 987
 988 void NETMOD::exit_sequence()
 989 {
 990  NETlink *pomlink;
 991
 992  ::close(kernel_sock);
 993  ::close(listen_sock);
 994  unlink(kername);
 995  pomlink = Links.first();
 996  while (pomlink!=NULL)
 997  {
 998   ::close(pomlink->sock);
 999   pomlink=Links.next();
1000   }
1001  exit(0);
1002 }
1003
1004 void NETMOD::disconnect_seq()
1005 {
1006  MESSAGE m;
1007  NETlink *p;
1008
1009  bzero(&m,sizeof(MESSAGE));
1010  m.msg_type = MSG_NET;
1011  m.param.pword[0] = NET_DISCONNECT;
1012  m.param.pword[1] = MyNode;
1013
1014  p=Links.first();
1015  while(p!=NULL)
1016  {
1017   send_to_node(p,&m);
1018   p=Links.next();
1019  }
1020  p=Links.first();
1021  while(p!=NULL)
1022  {
1023   ::close(p->sock);
1024   p=Links.next();
1025  }
1026  Links.clear();
1027 }
1028
1029 NETlink *NETMOD::findNETlink(int node)
1030 {
1031  NETlink *pomlink;
1032  pomlink=Links.first();
1033  while(pomlink!=NULL)
1034  {
1035   if (pomlink->node_number == node) return(pomlink);
1036   pomlink=Links.next();
1037  }
1038  return(pomlink);
1039 }
1040
1041 INTlink *NETMOD::findINTlink(int id)
1042 {
1043  INTlink *pomlink;
1044  pomlink=Interpreters.first();
1045  while(pomlink!=NULL)
1046  {
1047   if (pomlink->ID == id) return(pomlink);
1048   pomlink=Interpreters.next();
1049  }
1050  return(pomlink);
1051 }
1052
1053
1054 /* ----------------    Sending code to a remote node -------------- */
1055
1056 void NETMOD::transmit_file(int ton, char *fname, int fromINT)
1057 {
1058  FILE *f;
1059  MESSAGE msg;
1060  char fn[80];
1061  char b[255];
1062  unsigned char buffer[FILE_BUFFER_SIZE];
1063  protdescr proto;
1064  int i,tsock,sock;
1065  unsigned int sz;
1066  NETlink *outlink;
1067  struct sockaddr_in svr;
1068  fd_set rset,wset;
1069
1070
1071 // **************** CCD FILE
1072
1073  strcpy(fn,fname);
1074  strcat(fn,".ccd");
1075  f = fopen(fn,"rb");
1076  if (f!=NULL)
1077  {
1078  fseek(f,0,SEEK_END);
1079  msg.param.pword[1] = ftell(f);
1080  fclose(f);
1081  f = fopen(fn,"rb");
1082
1083  strcpy(b,rindex(fname,'/'));
1084  for(i=0;i<strlen(b);i++)
1085    b[i] = b[i+1];
1086  msg.param.pword[0] = NET_CCD_START;
1087  strcpy(msg.param.pstr,b);
1088
1089  outlink = findNETlink(ton);
1090  if (outlink==NULL) exit(1);
1091  bzero(&svr,sizeof(svr));
1092  sock = socket(AF_INET,SOCK_STREAM,0);
1093  svr.sin_family = AF_INET;
1094  svr.sin_addr.s_addr = INADDR_ANY;
1095  svr.sin_port = htons(CODEPORT);
1096  bind(sock, (struct sockaddr*)&svr, sizeof(svr));
1097  listen(sock,5);
1098  send_to_node(outlink, &msg);
1099  sz=sizeof(svr);
1100  FD_ZERO(&rset);FD_ZERO(&wset);
1101  FD_SET(sock,&rset);
1102  if (select(sock+1,&rset,&wset,0,0))
1103   if (FD_ISSET(sock,&rset))
1104     tsock = accept(sock, (struct sockaddr*)&svr,&sz );
1105  if (tsock>0)
1106  {
1107   close(sock);
1108   while (!feof(f))
1109    {
1110     i = fread(&buffer,1,sizeof(buffer),f);
1111     write(tsock,&buffer,i);
1112     FD_ZERO(&rset);FD_ZERO(&wset);
1113     FD_SET(tsock,&wset);
1114     select(tsock+1,&rset,&wset,0,0);
1115     }
1116   close(tsock);
1117  }
1118  fclose(f);
1119  } // f!=NULL
1120   else
1121      {
1122     sprintf(b,"Cannot open file to send %s\n",fname);
1123     write_at_console(b);
1124    }
1125
1126
1127 // *************** PCD FILE
1128
1129  strcpy(fn,fname);
1130  strcat(fn,".pcd");
1131  f = fopen(fn,"r");
1132   if (f!=NULL)
1133  {
1134  fseek(f,0,SEEK_END);
1135  msg.param.pword[1] = ftell(f);
1136  fclose(f);
1137  f = fopen(fn,"rb");
1138
1139  strcpy(b,rindex(fname,'/'));
1140  for(i=0;i<strlen(b);i++)
1141    b[i] = b[i+1];
1142  msg.param.pword[0] = NET_PCD_START;
1143  strcpy(msg.param.pstr,b);
1144
1145  outlink = findNETlink(ton);
1146  if (outlink==NULL) exit(1);
1147  bzero(&svr,sizeof(svr));
1148  sock = socket(AF_INET,SOCK_STREAM,0);
1149  svr.sin_family = AF_INET;
1150  svr.sin_addr.s_addr = INADDR_ANY;
1151  svr.sin_port = htons(CODEPORT1);
1152  bind(sock, (struct sockaddr*)&svr, sizeof(svr));
1153  listen(sock,5);
1154  send_to_node(outlink, &msg);
1155  sz=sizeof(svr);
1156  FD_ZERO(&rset);FD_ZERO(&wset);
1157  FD_SET(sock,&rset);
1158  if (select(sock+1,&rset,&wset,0,0))
1159   if (FD_ISSET(sock,&rset))
1160     tsock = accept(sock, (struct sockaddr*)&svr,&sz );
1161  if (tsock>0)
1162  {
1163   close(sock);
1164   while (!feof(f))
1165    {
1166     i = fread(&proto,1,sizeof(proto),f);
1167     write(tsock,&proto,i);
1168     FD_ZERO(&rset);FD_ZERO(&wset);
1169     FD_SET(tsock,&wset);
1170     select(tsock+1,&rset,&wset,0,0);
1171     }
1172   close(tsock);
1173  }
1174  fclose(f);
1175  } // f!=NULL
1176   else
1177    {
1178     sprintf(b,"Cannot open file to send %s\n",fname);
1179     write_at_console(b);
1180    }
1181
1182
1183  msg.msg_type = MSG_NET;
1184  msg.param.pword[0] = NET_TRANSMITTED;
1185  msg.param.pword[1] = fromINT;
1186  send_to_kernel(&msg);
1187
1188
1189 }
1190
1191
1192
1193 void NETMOD::conn_info(int sk)
1194 {
1195  NETlink *pom;
1196  MESSAGE m;
1197  int k=0;
1198  char poms[255];
1199
1200  m.msg_type = MSG_NET;
1201  m.param.pword[0] = NET_INFO;
1202  strcpy(m.param.pstr,"");
1203  pom=Links.first();
1204  while (pom!=NULL)
1205  {
1206   sprintf(poms,"%d=%s;",pom->node_number,pom->addr);
1207   strcat(m.param.pstr,poms);
1208   k++;
1209   if (k==12)
1210   {
1211     m.param.pword[1]=12;
1212     write(sk,&m,sizeof(MESSAGE));
1213     k=0;
1214    }
1215   pom=Links.next();
1216  }
1217  if (k>0)
1218   {
1219    m.param.pword[1]=k;
1220    write(sk,&m,sizeof(MESSAGE));
1221    }
1222   m.msg_type = MSG_NET;
1223   m.param.pword[0] = NET_INFO_END;
1224   write(sk,&m,sizeof(MESSAGE));
1225 }
1226
1227 int main(int argc,char **argv)
1228 {
1229  NETMOD netter(argv[1]);
1230  netter.run();
1231  return 0;
1232 }