Multi-threading with ports
<p>I am new to multi-threading and I'm trying to create a network where multiple nodes try to talk to each other through various unique ports, all on the same computer. Each node is assigned multiple Tx and Rx port numbers depending on how many nodes it is connected to and every Tx port on a node maps to just one Rx port on some other node. So for e.g., if I have 2 connected nodes A and B, then my port maps would look like: at A , tx_port=4000 rx_port=4100 destination_rx_port=4101; at B, tx_port=4002 rx_port=4101 dest_rx_port=4100. This list would get longer at each node as it connects with more nodes. </p>

<p>Now on every node, I create as many threads as there are receiving ports to listen to all of them simultaneously. I then send a HELLO from each node to all of its connections and keep sending it until an acknowledgment is heard from each of them. Now coming back to the case of A and B, if I run A first and then B, B's HELLO is heard by A and is also acknowledged, and B stops its HELLO transmission. But A's HELLO is never heard by B and B always crashes (aborts). The situation is the reverse when I run B first and then A.</p>

<p>But I found out that if I din't create the threads and just placed my port_listen() function inside main under a while(1) loop then B is able to listen A's HELLO. The problem appears to be that I have to create the thread to listen on a port before I send anything to that port, but that sounds silly and I'm sure there is a better explanation. I would be grateful if anyone helped me. I haven't posted my code because it is rather big and messy for now, and has a lot of other things going on but as I've checked, the code gives the same problem even after commenting them out.</p>

<p>Thanks.</p>

<pre><code>#include "common.h"
#include "newport.h"
#include&lt;iostream&gt;
#include&lt;fstream&gt;
#include&lt;string.h&gt;
#include&lt;stdlib.h&gt;


using namespace std;


Address *my_rx[50];
Address *my_tx[50];
Address *dest[50];
LossyReceivingPort *my_rx_p[50];
mySendingPort *my_p[50];

pthread_t rec_threads2[1000];    

int tx_ports[100];
int rx_ports[100];
int dest_rx_ports[100];
char node_id[100];
char dest_id[100];

int rx_port_count=0;
int nodest;
char tempc, rec_dest;   
int distinct=0; // distinct would become 0-9 =10 - (argc-1)
int w;





struct tab // only stores number of connections for each node
{int count; char node_id;
void init(){ count=1;}
};

struct neighbor // for each different link, therefore packet number can be accessed from here i guess
{ char neighbor; int count; char self;  
void init(){count=0;neighbor='\0';}
};

struct routing_table{
   char dest[100];
   int hop_count[100];
   char next_hop[100];
   void init(){for (int i=0;i&lt;100;i++){hop_count[i]=0; next_hop[i]='\0';dest[i]='\0';}}
};  


 tab node[50];
 neighbor neighbor_table [50];
 routing_table info_at[50];



void *rec_thread(void *obj)
{  
    char num1,num2;
    Address *addr;
    int hold, fetch, index, packet_num;
    int key;

mySendingPort *my_p= (mySendingPort *)obj;

addr=my_p-&gt;getRemoteAddr(); 
fetch=addr-&gt;getPort();
for (int p=1;p&lt;=rx_port_count;p++)
if(fetch==dest_rx_ports[p])
hold=p;


while(1)
{

{    
     Packet *pAck;          
    pAck=my_rx_p[hold-1]-&gt;receivePacket();
        if(pAck!=NULL)
    {            
               num1=pAck-&gt;accessHeader()-&gt;getOctet(1);
               my_p-&gt;global_flag=0;

           switch(num1)
           {
                       case 'H':{ packet_num=pAck-&gt;accessHeader()-&gt;getShortIntegerInfo(4); // get the packet number;

                                  Packet * my_ack_packet;        
                                  my_ack_packet = new Packet();      
                                  my_ack_packet-&gt;setPayloadSize(0);
                                  PacketHdr *hdr = my_ack_packet-&gt;accessHeader();                                      

                                 num2=pAck-&gt;accessHeader()-&gt;getOctet(2); // getting the destination ID to use as source ID.                                       
                                 hdr-&gt;setOctet(num2,0);// setting up the source address of the ACK
                                 hdr-&gt;setOctet('A',1);//packet type is Ack
                                 hdr-&gt;setOctet('H',2);//ack type is hello
                                 hdr-&gt;setShortIntegerInfo(packet_num,3);// copying the packet number
                                 num2=pAck-&gt;accessHeader()-&gt;getOctet(0);
                                 hdr-&gt;setOctet(num2,5); // setting up the destination ID
                                 my_p-&gt;sendPacket(my_ack_packet);     

                                 cout &lt;&lt; " node "&lt;&lt; node_id[hold]&lt;&lt;" receiving a HELLO from node " &lt;&lt; num2 &lt;&lt; "...sending ACK" &lt;&lt; endl; 
                                 delete my_ack_packet;
                                 if(packet_num&gt; my_rx_p[hold-1]-&gt;last_rec_packet_num)                                     
                                 { 
                                   num2=pAck-&gt;accessHeader()-&gt;getOctet(3);
                                   if(num2=='r'||num2=='R')
                                    { num2=pAck-&gt;accessHeader()-&gt;getOctet(0);

                                    if(num2!=neighbor_table[hold-1].neighbor)
                                    {
                                    neighbor_table[hold-1].neighbor=num2;
                                    neighbor_table[hold-1].count++;
                                    } 

                                    }

                                 }       // end packet num &gt; last rec                                                                                                     

                                 my_rx_p[hold-1]-&gt;last_rec_packet_num=packet_num;

                                 }


                                 break;                         
                       case 'A':{num2=pAck-&gt;accessHeader()-&gt;getOctet(2);


                                 switch(num2)
                                 {case 'H':my_p-&gt;timer_h.id='H';my_p-&gt;setACKflag(true,'H');my_p-&gt;timer_h.stopTimer();

                                           delete my_p-&gt;lastPkt_h;

                                           break;
                                  case 'D':my_p-&gt;timer_d.id='D';my_p-&gt;setACKflag(true,'D');my_p-&gt;timer_d.stopTimer(); 

                                           delete my_p-&gt;lastPkt_d;
                                           break;
                                  case 'U':my_p-&gt;timer_u.id='U';

                                            delete my_p-&gt;lastPkt_u;
                                           my_p-&gt;setACKflag(true,'U');
                                           my_p-&gt;timer_u.stopTimer();
                                           break;
                                } 
                                 num2=pAck-&gt;accessHeader()-&gt;getOctet(0);
                                 cout &lt;&lt; "\nThe "&lt;&lt;pAck-&gt;accessHeader()-&gt;getOctet(2)&lt;&lt;" packet has been acknowledged at node " &lt;&lt; node_id[hold]&lt;&lt;" by node "&lt;&lt;num2 &lt;&lt;endl;                                       
                                }                                                                            
           }// end switch                 


 }  // end if


 else cout&lt;&lt;"\n no packet received";     

 }
 } // end while
}                     





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


ifstream fin;
FILE *f=fopen("config.txt","r");
char str[1000];
int j, index, fields; 
char temp[1000];

const char* hname = "localhost";       
int count;
int thread_count=0, fetch; 

while(!feof(f))
{


               if(fgets(str,1000,f)!=NULL)
               {

                              j=0;            
                              while(str[j]!=' ')
                              {                       
                                      temp[j]=str[j];      
                                      j++;              
                              }
                              temp[j]='\0';
                              fields=0;
                              j=0;

                              for (int i=1;i&lt;argc;i++) // change made
                              {

                              if(strcmp(argv[i],temp)==0)
                              {                          

                                                        while(str[j]!='\0')
                                                       {   

                                                                        index=0;    

                                                                      while((str[j]!=' ') &amp;&amp; (str[j]!='\0') &amp;&amp; (str[j]!='\n'))

                                                                       {

                                                                                       temp[index]=str[j];

                                                                                       j++; 
                                                                                       index++;


                                                                      }
                                                                       if(str[j]!='\0')
                                                                       {j++;}


                                                                           fields++;
                                                                           temp[index]='\0';

                                                                       switch(fields)
                                                                       {               case 1:{rx_port_count++;
                                                                                               node_id[rx_port_count]=temp[0];}break;

                                                                                       case 2: {
                                                                                                rx_ports[rx_port_count]=short(atoi(temp));break;

                                                                                                } 
                                                                                       case 3: {
                                                                                                tx_ports[rx_port_count]=short(atoi(temp));break;
                                                                                               }
                                                                                       case 4: {dest_rx_ports[rx_port_count]=short(atoi(temp));break;

                                                                                                }
                                                                                       case 5: dest_id[rx_port_count]=temp[0];         
                                                                       } //end switch


                                                     }//end (str[j]!='\0')


                            } //end (strcmp(argv[i],temp)==0)
                            }//end (int i=1;i&lt;=argc;i++)
             }//end (fgets(str,1000,f)!=NULL)
} //end (!feof(f))
    cout&lt;&lt;rx_port_count&lt;&lt;'\n';          

for (int i=1;i&lt;=rx_port_count;i++)
{       
    cout&lt;&lt;rx_ports[i]&lt;&lt;tx_ports[i]&lt;&lt;dest_rx_ports[i]&lt;&lt;'\n';  

    my_rx[i-1]=new Address("localhost", rx_ports[i]);

    my_rx_p[i-1]=new LossyReceivingPort(0.2);
    my_rx_p[i-1]-&gt;setAddress(my_rx[i-1]);
    my_rx_p[i-1] -&gt;init();

    my_tx[i-1]=new Address(hname,tx_ports[i]); // no here lies the problem

    dest[i-1]=new Address("localhost", dest_rx_ports[i]); // here lies the problem

    my_p[i-1] = new mySendingPort();
    my_p[i-1]-&gt;setAddress(my_tx[i-1]);
    my_p[i-1]-&gt;setRemoteAddress(dest[i-1]);
    my_p[i-1]-&gt;init();     
    my_p[i-1]-&gt;thread_count=thread_count;   
    my_p[i-1]-&gt;ack_init();


}// end (int i=1;i&lt;=rx_port_count;i++)

node[0].node_id=node_id[1];
for (j=0;j&lt;50;j++)
{node[j].init();neighbor_table[j].init();info_at[j].init();}

for (j=2;j&lt;=rx_port_count;j++)
{
if(node_id[j]!=node_id[j-1])
{ distinct++;node[distinct].node_id=node_id[j];
}

 else
{
node[distinct].count++;}

}

for (j=0;j&lt;=distinct;j++)
cout&lt;&lt;node[j].node_id&lt;&lt;node[j].count&lt;&lt;'\n';


    // H E L L O 
index=0;        
count=0;

 for (int i=1;i&lt;=rx_port_count;i++)
{int error=pthread_create(&amp;rec_threads2[i],NULL,&amp;rec_thread, (void *)my_p[i-1]);
 if(error)
 cout&lt;&lt;"\nreceiver thread creation failed\n";
} 





for(j=0;j&lt;=distinct;j++)
{                   
for ( int i=1;i&lt;=node[j].count;i++)
{       index++;      

Packet * my_packet= new Packet();    
my_p[index-1]-&gt;packet_num++;
my_packet-&gt;setPayloadSize(0);
    PacketHdr *hdr = my_packet-&gt;accessHeader();

    hdr-&gt;setOctet(node_id[index],0); // sender's address
    hdr-&gt;setOctet('H',1);           // packet type
    hdr-&gt;setOctet(dest_id[index],2);  // destination address                             
    hdr-&gt;setOctet('H',3);   // device type        
    hdr-&gt;setShortIntegerInfo(my_p[index-1]-&gt;packet_num,4);// packet number

    my_p[index-1]-&gt;timer_h.thread_count=my_p[index-1]-&gt;thread_count;

    my_p[index-1]-&gt;sendPacket(my_packet);

    my_p[index-1]-&gt;lastPkt_h=new Packet();
    my_p[index-1]-&gt;lastPkt_h-&gt;copy(my_packet);                         
    my_p[index-1]-&gt;setACKflag(false,'H'); // change made
    my_p[index-1]-&gt;timer_h.id='H';// change made     
    my_p[index-1]-&gt;timer_h.startTimer(2.5);  // change made

    delete my_packet;




}// end (int i=1;i&lt;=rx_port_count;i++)      
} // end j=0;j&lt;=distinct;j++)


while(1){}


/*while(1)
{ //cout&lt;&lt;"\n coming here";
 p=my_rx_p[0]-&gt;receivePacket();
 if(p!=NULL)
 cout&lt;&lt;"\npacket received ";
 else cout&lt;&lt;"\n no packet ";
} */

} // end try
catch (const char *reason ) {
    cerr &lt;&lt; "Exception:" &lt;&lt; reason &lt;&lt; endl;
    exit(-1);
  }  

}  //end main 
</code></pre>
