queue.cc

initial working phase of the design of said editor, featuring multicasting, advanced linux keyboard

#include "queue.h"

//#include <stdlib.h>			// malloc, free
#include <stdio.h>			// stdin, fileno
#include <unistd.h>			// read

char message[ 1500 ]; // dynamic allocation segfaults below

queue::~queue() // set semaphore?, notify of absence
{
   event* temp;
   for(; q != NULL; q = temp )
   {
      temp = q -> next;
      delete q;
   }
//   free( lastfrom );
   delete lastfrom;
}

#include <sys/time.h>			// timeval
#include <time.h>			// gettimeofday
queue::queue( term* _output, int _sock, struct sockaddr_in _to, int id )
: sock( _sock ), to( _to )
{
   locked = false;
   q = NULL;
   output = _output;
   clients = 1;
//   lastfrom = ( long* )malloc( sizeof( long ) * 4 );
   lastfrom = new long[ 4 ];
   lastfrom[ 0 ] = id;
   struct timezone tz;
   gettimeofday( &when, &tz );
   lastfrom[ 1 ] = when.tv_sec;
   lastfrom[ 2 ] = when.tv_usec;
   lastfrom[ 3 ] = 0; // cursor
   strokes = 0;
   sending = 0;
}

#include "vector.h" // vector::mcast
#include "thread.h" // spawn

void queue::signal( char* buffer, int length, int sender, timeval _stamp )
{
   // socket: assumes message is a remote queue
   // message could also be: to leave
   {
      int it = 0, c = 0;

   timeval lcd;
   event* temp;
   for(; it < length; it++ )
   {
      temp = NULL;
      event* n = new event;
      memcpy( &( n -> tv ), ( it * 16 ) + buffer, sizeof( timeval ) );
//   printf( "%08X %08X\r\n", n -> tv.tv_sec, n -> tv.tv_usec );
//   fflush( stdout );
      memcpy( &( n -> input ), ( it * 16 ) + buffer + sizeof( timeval ),
	 sizeof( int ) );
      memcpy( &( n -> owner ), ( it * 16 ) + buffer + sizeof( timeval )
	 + sizeof( int ), sizeof( int ) );
      n -> next = NULL;

      if( n -> tv.tv_sec < when.tv_sec
	 || ( n -> tv.tv_sec == when.tv_sec
	    && n -> tv.tv_usec <= when.tv_usec ) )
      {
	 delete n; // before the queue is irrelevant
//	 output -> pipe( 0, '_' ); // this happens a lot
/*
   if( it > 0 )
   {
      printf( "\r\n%d %c", it, n -> input );
      fflush( stdout );
   }
*/
	 continue;
      }
      if( !q )
      {
	 temp = q = n;
      }
      else if( q -> tv.tv_sec > n -> tv.tv_sec
	 || ( q -> tv.tv_sec == n -> tv.tv_sec
	    && q -> tv.tv_usec > n -> tv.tv_usec ) )
      {
	 n -> next = q;
	 temp = q = n;
      }
      else if( q -> tv.tv_sec == n -> tv.tv_sec
	 && q -> tv.tv_sec == n -> tv.tv_usec )
      {
	 delete n;
	 continue;
      }
      else // temp never reaches beyond the last given timestamp
      {
	 // this loop assumes the queue is in-order, it is
	 for( temp = q; temp -> next != NULL
	    && ( temp -> next -> tv.tv_sec < n -> tv.tv_sec
	       || ( temp -> next -> tv.tv_sec == n -> tv.tv_sec
		  && temp -> next -> tv.tv_usec < n -> tv.tv_usec ) );
	    temp = temp -> next );

	 if( temp -> next != NULL
	    && ( temp -> next -> tv.tv_sec == n -> tv.tv_sec
	       && temp -> next -> tv.tv_sec == n -> tv.tv_usec ) )
	 {
	    delete n;
	    continue;
	 }
	 n -> next = temp -> next;
	 temp = temp -> next = n;
      }
      if( temp == NULL )
      {
//	 output -> beep();
	 continue;
//	 return;
      }
//      printf( "x %08X %08X\r\n", temp -> tv.tv_sec, temp -> tv.tv_usec );
//      printf( "o %08X %08X\r\n", _stamp.tv_sec, _stamp.tv_usec );
//      fflush( stdout );

      int cl = 0;
      for(; cl < clients && lastfrom[ cl * 4 ] != ( u_long )sender; cl++ );
      if( cl < clients )
      {
//	 lastfrom[ cl * 4 + 1 ] = temp -> tv.tv_sec;
//	 lastfrom[ cl * 4 + 2 ] = temp -> tv.tv_usec;
	 if( lastfrom[ cl * 4 + 1 ] < _stamp.tv_sec
	 || ( lastfrom[ cl * 4 + 1 ] == _stamp.tv_sec
	    && lastfrom[ cl * 4 + 2 ] < _stamp.tv_usec ) )
	 {
	    lastfrom[ cl * 4 + 1 ] = _stamp.tv_sec;
	    lastfrom[ cl * 4 + 2 ] = _stamp.tv_usec;
	 }
      }
      else // there should also be a way to remove a client
      {
	 clients++;
//	 long* n = ( long* )malloc( sizeof( long ) * clients * 4 );
	 long* n = new long[ clients * 4 ];
	 memcpy( n, lastfrom, sizeof( long ) * ( clients - 1 ) * 4 );
//	 if( lastfrom != NULL ) free( lastfrom );
	 if( lastfrom != NULL ) delete lastfrom;
	 int* x = new int;
	 delete x;
	 lastfrom = n;
	 lastfrom[ cl * 4 ] = sender;
//	 lastfrom[ cl * 4 + 1 ] = temp -> tv.tv_sec;
//	 lastfrom[ cl * 4 + 2 ] = temp -> tv.tv_usec;
	 lastfrom[ cl * 4 + 1 ] = _stamp.tv_sec;
	 lastfrom[ cl * 4 + 2 ] = _stamp.tv_usec;
	 lastfrom[ cl * 4 + 3 ] = output -> add_cursor();
      }
      lcd.tv_sec = 0x7FFFFFFF;
      lcd.tv_usec = 1000000;
      ( clients > 1 ) ? cl = 1 : cl = 0;
      for(; cl < clients; cl++ ) // skip local timestamp
      {
	 if( lastfrom[ cl * 4 + 1 ] < lcd.tv_sec )
	 {
	    lcd.tv_sec = lastfrom[ cl * 4 + 1 ];
	    lcd.tv_usec = lastfrom[ cl * 4 + 2 ];
	 }
	 else if( lastfrom[ cl * 4 + 1 ] == lcd.tv_sec
	       && lastfrom[ cl * 4 + 2 ] < lcd.tv_usec )
	 {
	    lcd.tv_usec = lastfrom[ cl * 4 + 2 ];
	 }
      }
// lcd should be the last time all data was collected from this host
      when = lcd; // unanimous update
   }

   if( sender != id )
   {
      locked++; // now sending the queue, don't want our copy
      c = 0;
      for( temp = q; temp != NULL; temp = temp -> next ) c++;
      if( c == 0 )
      {
	 locked--;
	 return;
      }
//      char* body,* message = ( char* )malloc( 20 + 16 * c );
      char* body; //,* message = new char[ 20 + 16 * c ];

      sprintf( message, "* %08X %08X\n", c, lastfrom[ 0 ] );

      body = message + 32;
      int strings = 0;
      for( temp = q, it = 0; it < c; temp = temp -> next, it++ )
      {
	 if( temp -> input & 0x80000000 == 0x80000000 )
//	 if( temp -> cx != NULL )
	 {
	    strings++;
//	    memcpy( body, temp -> cx, strlen( temp -> cx ) + 1 );
//	    body += ???
	 }
      }
      *( int* )( message + 20 ) = 0; //strings; // no string table
      *( timeval* )( message + 24 ) = when;

      for( temp = q, it = 0; it < c; temp = temp -> next, it++ )
      {
	 memcpy( ( 16 * it ) + body, &( temp -> tv ), sizeof( timeval ) );
	 memcpy( ( 16 * it ) + body + sizeof( timeval ),
	    &( temp -> input ), sizeof( int ) );
	 memcpy( ( 16 * it ) + body + sizeof( timeval ) + sizeof( int ),
	    &( temp -> owner ), sizeof( int ) );
      }
//   printf( ",<%d>", 32 + 16 * c );
//   fflush( stdout );
      sendto( sock, message, 32 + 16 * c, 0,
	 ( struct sockaddr* )&to, sizeof( to ) ); // repeated message
      locked--;
   }
//      free( message );
//      delete message;
   }
   spawn( &vector::mcast, NULL );
}

void queue::disperse()
{
      for(; q != NULL && ( q -> tv.tv_sec < when.tv_sec //lcd.tv_sec
	 || ( q -> tv.tv_sec == when.tv_sec
	    && q -> tv.tv_usec <= when.tv_usec ) ); ) //q = q -> next )
      {
	 // decode q -> input, using output
	 int cursor = 0;
	 for(; cursor < clients
	 && lastfrom[ 4 * cursor ] != q -> owner; cursor++ );
	 if( cursor < clients ) cursor = lastfrom[ 4 * cursor + 3 ];
	 else exit( 1 );

	 int flags = q -> input & 0xFFFFFF00;
	 q -> input &= 0x000000FF;
	 switch( flags )
	 {
	 case 0x00000000:
	    output -> pipe( cursor, ( char )( q -> input ) );
	    break;
	 case 0x00000100:
	    output -> func( cursor, ( char )( q -> input ), 0 );
	    break;
	 case 0x00000200:
	    output -> arrow( cursor, ( char )( q -> input ) );
	    break;
	 case 0x00000300:
	    switch( ( char )( q -> input ) )
	    {
	    case 1:
	       output -> bs( cursor );
	       break;
	    case 2:
	       output -> cr( cursor );
	       break;
	    case 3:
	       output -> del( cursor );
	       break;
	    case 4:
	       output -> ins( cursor );
	       break;
	    case 5:
	       output -> home( cursor );
	       break;
	    case 6:
	       output -> end( cursor );
	       break;
	    case 7: // flags not distinguished here
	       output -> pan( cursor, 4, false );
	       break;
	    case 8:
	       output -> pan( cursor, 6, false );
	       break;
	    }
	    break;
	 }
	 event* old = q;
	 q = q -> next;
	 delete old;
      }
}

// how shall kbd have known, flagged locally we were changing cx ?

void queue::enqueue( u_int input )
{
   word limit = 16;

   if( sending++ == 0 )
   {
      inbuf = new char[ 32 + 16 * limit ];
   }
   else if( sending > limit )
   {
      // reallocate and copy
      //inbuf = new char[ 32 + 16 * limit ];
      --sending;
      return;
   }
   ++strokes; // this gets called from kbd, which will decrement strokes

//   printf( "enqueue " );
//   fflush( stdout );

   char* pos = inbuf + 32 + ( ( sending - 1 ) * 16 ); //strlen( inbuf );
   // if stamp is same, increment it
   // this assumes it is called in the same order as calls to vector::key
   memcpy( pos, ( void* )&stamp, sizeof( struct timeval ) );
   pos += sizeof( struct timeval );
   memcpy( pos, ( void* )&input, sizeof( int ) );
   pos += sizeof( int );
   memcpy( pos, ( void* )&id, sizeof( int ) );
   pos += sizeof( int );
}

/*
input: strokes++
(enqueue: sending++, ++strokes)
kbd: strokes--
vector::key: xmit: --strokes
*/

void queue::xmit()
{
   if( strokes == 0 || sending == 0 ) return;
   // this is happening every time
   if( --strokes > 0 ) // need a special case indicating no more input
   {
//      output -> pipe( 0, '&' );
//      printf( "%d\r\n", strokes );
      return;
   }
// now put a timestamp on the whole message

   sprintf( inbuf, "* %08X %08X\n", sending, id );
   int strings = 0;
   memcpy( inbuf + 20, ( void* )&strings, sizeof( int ) );

//   if( sending > 1 ) output -> pipe( 0, '+' );

   if( clients > 1 ) // this message won't be recieved locally
   {
      lastfrom[ 1 ] = stamp.tv_sec;
      lastfrom[ 2 ] = stamp.tv_usec;
   }
   memcpy( inbuf + 24, ( void* )&stamp, sizeof( timeval ) );
   sendto( sock, inbuf, 32 + 16 * sending, 0,
//   sendto( sock, inbuf, 24 + 16 * sending, 0,
      ( struct sockaddr* )&to, sizeof( to ) ); // local transmission

   delete[] inbuf;

   sending = 0;
}

void queue::pipe( u_char input )
{
   enqueue( ( u_int )input );
}

void queue::func( u_char input )
{
   enqueue( ( u_int )input | 0x00000100 );
}

void queue::arrow( u_char input )
{
   enqueue( ( u_int )input | 0x00000200 );
}