Cprocess_clients.cc

Go to the documentation of this file.

extern "C"
{
  #include <sys/times.h>
}

static inline double difftimes(clock_t t1, clock_t t0)
{
  // refer "man 2 times"
  return static_cast<double>(t1-t0)/static_cast<double>(sysconf(_SC_CLK_TCK));
}


void Cprocess_clients::process_data_stream_ecm(const connection_waiter& my_connection_waiter)
{
  streampos fpos = Recovery_from_file.tellg();
  int Number_of_curves = 0;
  string s;

  srand(time(NULL)); // initialize random-seed (time-function should be random enough for our purpose)

  Recovery_from_file >> Number_of_curves;
  Recovery_from_file.ignore(1,'\n'); // read over eol

  while (Number_of_curves<elcu_Kurven)
  {
    unix_io_stream daten(my_connection_waiter); // wait for connection and connect to stream

    daten >> s;

    if (s == "ECM?")
     {
       int ecm_sigma = rand();
       daten << ecm_sigma << " "
             << elcu_Phase1 << " "
             << elcu_Phase2 << " "
             << n << endl; // transmit sigma and n

       Number_of_curves++;
       cout_network << "elliptic curve #" << Number_of_curves << "/" << elcu_Kurven
            << " with sigma=" << ecm_sigma
            << " sent to " << peer_info(daten) << endl;

       // write number of finished curves to recovery-file...
       Recovery_to_file.seekp(fpos); 
       Recovery_to_file << Number_of_curves << endl << flush;
       ecm_curves_processed=Number_of_curves;

       daten << flush;
       continue;
     }

    if (s == "Faktor(ECM)!") // accept factors
      {
        cout_network << "Factor received from " << peer_info(daten) << endl;
        int sigma;
        mpz_t x;
        mpz_init(x);
        daten >> sigma >> x; // get info about curve (sigma) and get factor
        while (isspace(daten.peek())) daten.get();
        string ecm_s = "ecm";
        if (daten.peek()!=EOF) daten >> ecm_s;
        mpz_gcd(x,x,n); // compute gcd (it may happen, that the factor has been already discovered!)
        if (mpz_cmp_ui(x,1)!=0)
         {
           // theoretically the factor can divide multiple times...
           const unsigned int exponent = mpz_remove(n,n,x);
           if (mpz_probab_prime_p(x,probab_prime_checks))
             {
               cout << x << " is factor." << endl;
               if (mpz_sizeinbase(x,10)<28)
                {
                  ostringstream comment;
                  comment << " [" << ecm_s << "]";
                  Factorization_to_file << MAL(x,exponent,comment) << flush;
                }
               else
                {
                  ostringstream comment;
                  comment << " [" << ecm_s << ",sigma=" << sigma << "]";
                  Factorization_to_file << MAL(x,exponent,comment) << flush;
                }
             }
           else
             {
               cout << x << " is a composite factor." << endl;
               if (mpz_probab_prime_p(n,probab_prime_checks))
                {
                  if (exponent>1) mpz_pow_ui(x,x,exponent);
                  mpz_swap(n,x);
                  cout << x << " is factor. (factorswap)" << endl;
                  ostringstream comment;
                  comment << " [" << ecm_s << "/factorswap,sigma=" << sigma << "]";
                  Factorization_to_file << MAL(x,comment) << flush;
                }
               else
                {
                  if (mpz_sizeinbase(x,10)<28)
                   {
                     ostringstream comment;
                     comment << " [" << ecm_s << "] [composite]";
                     Factorization_to_file << MAL(x,exponent,comment) << flush;
                   }
                  else
                   {
                     ostringstream comment;
                     comment << " [" << ecm_s << ",sigma=" << sigma << "] [composite]";
                     Factorization_to_file << MAL(x,exponent,comment) << flush;
                   }
                }
             }
         }
        mpz_clear(x);
        daten << flush;

        if (mpz_probab_prime_p(n,probab_prime_checks))
          {
            Factorization_to_file << MAL(n);
            cout << "remaining factor: " << n << endl;
            cout << "remaining factor is most probably prime!" << endl;
            mpz_set_ui(n,1); // n ist faktorisiert
          }

        if ( (mpz_cmp_ui(n,1)==0) || Potenztest(n) ) // factorization complete?
          { 
            cout << "factorization successfully completed." << endl;
            Factorization_to_file << endl;
            ExitManager::StopFactorization(); // exit program (success!) and do final work...
          }
        tune_parameters(mpz_sizeinbase(n,10)); // as the name says: tune parameters

        // adapt changes to the recovery-file!
        // number to factorize has lowered!
        Recovery_buffer.close(); // close file
        Recovery_buffer.open(RecoveryFile.c_str(),ios::in|ios::out|ios::trunc); // re-open it (creating an empty file!)
        Recovery_to_file << n << endl; // write out new number to factorize
        fpos = Recovery_from_file.tellg(); // mark fileposition
        Recovery_to_file << Number_of_curves << endl << flush; // store number of finished curves (ecm)
        continue;
      }

    if (s == QsieveLogon)
     {
       // requesting MPQS parameters is not adequate while processing ecm!
       cerr << "Inadequate request of MPQS parameters from " << peer_info(daten) << endl;
       daten << -1 << " " << "please wait until ecm is processed!" << endl;
       continue;
     }

    cerr << "Invalid request \"" << s << "\""
         << " from " << peer_info(daten) << endl;
    daten << flush;
  }
  cout << "distributed ECM-Phase (server) completed." << endl;
}



// server action, when an invalid seek occurs in one of the combine methods
void Cprocess_clients::seek_emergency_handler(istream &in, const streampos pos)
{
  // istream is probably a read-only-stream for DynamicRelations::DynamicRelations_to_file;
  // for performance reasons we do not flush DynamicRelations_to_file after each write,
  // so it may occur, that this is causing the seek failure...
  // we can take appropriate action:

  cerr << "invalid seek position on istream, trying to resolve..." << endl;

  // first we try to get into the critical section to obtain the right to flush the ostream
  // if this fails, we wait few seconds and retry (maybe another thread has triggered flushing the ostream)

  in.clear();
  CCriticalSection CriticalSection(ServerMutex,false);
  if (CriticalSection.try_enter())
   {
     DynamicRelations::DynamicRelations_to_file.flush();
     CriticalSection.leave();
     in.seekg(pos); 
     if (in.peek()!=EOF) return; // good, problem is most probably resolved!
   }
  else
   {
     // since we're in a multithreaded environment, it may happen, that another thread flushes within
     // the next few seconds, so let's wait a little bit and retry
     sleep(2); in.seekg(pos); 
     if (in.peek()!=EOF) return; // good, problem is most probably resolved! (but it may still happen, that data is only written partially)
#if 1
     cerr << "invalid seek position on istream not resolved, trying to flush ostream without obtaining a lock" << endl;
     // since we have no chance to check whether we're inside (=the owner) of the critical section,
     // we flush the DynamicRelations::DynamicRelations_to_file without any locking...
     // if we're lucky, this works; if not, the server may crash :-(
     // cross fingers and give it a try ;-)
     in.clear(); DynamicRelations::DynamicRelations_to_file.flush(); in.seekg(pos);
     if (in.peek()!=EOF) return; // good, problem is most probably resolved!
#endif
    }
  // giving up:
  throw ios_base::failure("unresolvable invalid seek position on istream");
}

CCriticalSection::Mutex Cprocess_clients::ServerMutex;

void Cprocess_clients::process_data_stream(unix_io_stream &daten, const string client)
{
  string s;
  daten >> s;

  CCriticalSection CriticalSection(ServerMutex,false);
   // this function is a "critical section": since we are in an multithreaded environment,
   // we need to have control, who is entering and leaving this!
   // Only one thread at a time may be allowed to stay inside the critical section.
   // Permission to enter this section is controlled by a mutex mechanism.

  if (s == QsieveLogon)
    {
      cout_network << "Sending number to factor to " << client << endl;
      daten << n << " " << StaticFactorbase::Size() << " " 
            << Factor_Threshold << " " << LogicalSieveSize << endl;
      return;
    }

  if (s == "ECM?")
   {
     daten << "-1" << endl; // do not accept any elliptic curves requests anymore (-> sigma<0)
   }

  if (s == "Faktor(ECM)!") // but accept factors...
    {
      cout_network << "Factor received from " << client << endl;
      int sigma;
      mpz_t x;
      mpz_init(x);
      daten >> sigma >> x; // get info about curve (sigma) and get factor
      while (isspace(daten.peek())) daten.get();
      string ecm_s = "ecm";
      if (daten.peek()!=EOF) daten >> ecm_s;
      mpz_gcd(x,x,n); // compute gcd (it may happen, that the factor has been already discovered!)
      if (mpz_cmp_ui(x,1)==0) // does the "factor" still divide?
       { mpz_clear(x); return; } // "factor" doesn't divide (anymore)

      CriticalSection.enter();

      if (mpz_cmp_ui(x,1)!=0)
       {
         // theoretically the factor can divide multiple times...
         const unsigned int exponent = mpz_remove(n,n,x);
         if (mpz_probab_prime_p(x,probab_prime_checks))
           {
             cout << x << " is factor." << endl;
             if (mpz_sizeinbase(x,10)<28)
              {
                ostringstream comment;
                comment << " [" << ecm_s << "]";
                Factorization_to_file << MAL(x,exponent,comment) << flush;
              }
             else
              {
                ostringstream comment;
                comment << " [" << ecm_s << ",sigma=" << sigma << "]";
                Factorization_to_file << MAL(x,exponent,comment) << flush;
              }
           }
         else
           {
             cout << x << " is a composite factor." << endl;
             if (mpz_sizeinbase(x,10)<28)
              {
                ostringstream comment;
                comment << " [" << ecm_s << "] [composite]";
                Factorization_to_file << MAL(x,exponent,comment) << flush;
              }
             else
              {
                ostringstream comment;
                comment << " [" << ecm_s << ",sigma=" << sigma << "] [composite]";
                Factorization_to_file << MAL(x,exponent,comment) << flush;
              }
           }
       }
      mpz_clear(x);

      if (mpz_probab_prime_p(n,probab_prime_checks))
        {
          Factorization_to_file << MAL(n);
          cout << "remaining factor: " << n << endl;
          cout << "remaining factor is most probably prime!" << endl;
          mpz_set_ui(n,1); // n ist faktorisiert
        }

      if ( (mpz_cmp_ui(n,1)==0) || Potenztest(n) ) // factorization complete?
        { 
          cout << "factorization successfully completed." << endl;
          Factorization_to_file << endl;
          ExitManager::StopFactorization(); // exit program (success!) and do final work...
        }

      /* if program flow is at this position, then:
           A client (using ecm) has detected a new factor resulting in a
           partial factorization during the sieving phase. This leads to severe
           consequences:
            - We shouldn't ignore the new factor.
            - We cannot continue sieving.
            - abort the factorization and delete recovery data
              (-> recovery does not fit for the remaining number)
            - let the user restart with the remaining number.
      */
       {
          // we abort the factorization and print a remark,
          // that the remaining cofactor is still composite...

          Factorization_to_file << MAL(n) << " [composite]" << endl;
          cout << "remaining factor: " << n << endl
               << "remaining factor is composite!" << endl;
          cout << "factorization needs to be stopped to avoid inconsistent files." << endl
               << "please restart factorization with remaining number." << endl;

          Factorization_to_file << endl;
          ExitManager::StopFactorization(); // exit program (success)
        }
                 
      return;
    }

  if (s=="DynamicFactors?_ab_index")
    {
      // Online-clients should use this one to fetch dynamic factors.
      cout_network << "new-style dynamic factor request from " << client << endl;

      int i;
      daten >> i;
      const int start = i;

      cout_network << "Sending dynamic factors starting with index " << i
                   << " to " << client << endl;
      for (int j=0; j<20; ++j)
       { 
         const int i_start = i;
         while (i<DynamicFactorRelations.monitoredSize())
          {
            // binary transmission of data to increase throughput.
            // important: transmit data in a defined byteorder to be compatible!
            register unsigned int h=DynamicFactorRelations[i];
            char c[4];
            c[0]=h&0xff; c[1]=(h>>8)&0xff; c[2]=(h>>16)&0xff; c[3]=(h>>24)&0xff;
            daten.write(c,4); if (daten.fail()) break;
            //daten << DynamicFactorRelations[i] << " ";
            ++i;
          }
         //cerr << " ---- " << client << " ----  j = " << j << " -------" << endl;
         pthread_testcancel();
         if (i>i_start)
          { 
            daten.flush();
            j=0;
          }
         if (daten.fail()) break;
         sleep(2);
         pthread_testcancel();
       }
      //daten << "-1" << endl;
      if (daten.fail()) cout_network << "dynamic factors thread: transmission failure!" << endl;
      else
       {
         daten.put(0); daten.put(0); daten.put(0); daten.put(0); daten.flush();
       }
      cout_network << "Dynamic factors from " << start << " to " << i-1
                   << " sent to " << client << "." << endl;
      if (!daten.fail())
       {
         // read over remaining/redundant data
         for(int i=0; i<4; ++i)
          if (daten.peek()!=EOF)
           {
             int h=daten.get();
             cerr << "unexpected value (" << h << ") instead of EOF on stream!" << endl;
           }
       }
      cout_network << "Connection to " << client << " closed." << endl;
      return;
    }
  
  if (s=="DynamicFactors?_ab_fpos")
    {
      cout_network << "old-style dynamic factor request from " << client << endl;
      // this feature is deprecated!
      // Online-clients should use the above one to fetch dynamic factors.
      // However, for offline clients and downwards compatibility with
      // client-versions up to 2.92 this feature is still provided...

      //streampos startfpos; // this is meant
      long long int startfpos; // and this will compile...
      daten >> startfpos;

      CriticalSection.enter();

      ostringstream temp_stringstream; // temporary to avoid long process locks
      cout_network << "Generating dynamic factors starting with filepos " << startfpos << endl;
      const ostream::pos_type saved_fpos = DynamicRelations_to_file.tellp();
      DynamicRelations_to_file.seekp(0,ios::end);
      const ostream::pos_type fpos_to_transmit = DynamicRelations_to_file.tellp(); // use variable, because this is safer for exception handling
      DynamicRelations_to_file.seekp(saved_fpos);

#ifdef STL_STREAM_workaround
      daten << static_cast<streamoff>(fpos_to_transmit) << " ";
#else
      daten << fpos_to_transmit << " ";
#endif
      
      {
        CUnlockMutexAtDestruction SLP_Lock(DynamicFactorRelations.SLP_mutex);
        DynamicFactorRelations.SLP_mutex.lock();
        // this section can only proceed, if DynamicFactorRelations are locked!!
        // -> you should really try to avoid using deprecated protocol feature,
        //    as it is ugly slow!!   

        int prev=0; // transmit difference between numbers instead numbers theirselves
        for (TDynamicFactorRelations::const_iterator pos=DynamicFactorRelations.begin();
             pos != DynamicFactorRelations.end() && pos->sieveable(); ++pos)
#ifdef STL_STREAM_workaround
          if (static_cast<streamoff>((*pos).fpos)>=startfpos)
#else
          if ((*pos).fpos>=startfpos)
#endif
            {
              temp_stringstream << (*pos).factor-prev << " ";
              prev=(*pos).factor;
            }
      }

      const unsigned int KB = temp_stringstream.str().length()/1024;
      cout_network << "Sending dynamic factors starting with fileposition " << startfpos
                   << " (" << KB << "KB)" <<endl;

      CriticalSection.leave(); // allow other threads to sneak in

      if (!temp_stringstream.str().empty()) daten << temp_stringstream.str();
      if (daten.fail()) cout_network << "dynamic factors thread (old-style): transmission failure!" << endl;
      else daten << "-1" << endl;

      cout_network << "Connection (old-style) to " << client << " closed." << endl;
      return;
    }
  
  if (s=="Polynom?")
    {
      static CCriticalSection::Mutex myPolynomMutex;
      int intervall;
      daten >> intervall;
      cout_network << "Sending new MPQS interval (" << intervall << ")"
                   << " to " << client << endl;

      CCriticalSection CriticalPolynomSection(myPolynomMutex);
      // the critical section for serving MPQS polynomial data starts here...

      Polynom.save(daten); // Intervallanfang
      Polynom.compute_next_polynomial(intervall);
      Polynom.save(daten); // end of interval
      
      // for recovery:
      streampos fpos = Recovery_to_file.tellp();
      Polynom.save(Recovery_to_file); // save current mpqs polynomial
      Recovery_to_file.flush(); // be cautious...
      Recovery_to_file.seekp(fpos);
      return;
    }

  // LAZY-Variante (ersetzt die nunmehr obsolete BUSY-Variante,
  // die bis einschließlich Version 2.92 als Option noch implementiert war):
  // Wenn die empfangenen Relationen Korrekturfaktoren enthalten, werden
  // diese erst dann ausgewertet, wenn eine statische Relation entsteht.
  // Diese Variante ist insbesondere für das Client-Server-Modell
  // implementiert, da das relativ einfach ist und der Server alle
  // Relationen schnell entgegennehmen soll, was in der bisherigen
  // BUSY-Variante bei größeren Gleichungssystemen nicht mehr gewährleistet
  // werden konnte. Außerdem dürfte die LAZY-Variante Plattenplatz sparen.
  // Nachteil: Der Check auf defektierende Relationen wird nun
  // unverhältnismäßig aufwendig. (Per Induktion gilt zwar: Wenn jede Relation
  // getestet wird, dann reicht es, alle Faktoren (einschließlich der 
  // Korrekturfaktoren) und den quadrierten Restfaktor zusammenzumultiplizieren
  // und das Produkt modulo n auf 1 zu testen. Die Korrektheit der zu den
  // Korrekturfaktoren gehörigen Relationen ergibt sich dann durch einen
  // Existenzcheck der zugehörigen Relation (und der Induktionsannahme, dass
  // diese korrekt ist). Dieser Test ist also einfach zu implementieren, aber
  // während in der Busy-Variante der Test nicht allzu sehr ins Gewicht fällt,
  // nimmt er hier einen großen Teil der Verarbeitungszeit ein, da die
  // Relation nun nicht mehr als Zeichenkette verarbeitet werden kann, sondern
  // tatsächlich ausgewertet werden muß. Gerade aber die Auswertung sollte
  // durch die LAZY-Methode auf das notwendige Maß beschränkt werden.
  //
  // policy for LAZY-variant:
  //  1. receive relation as a string of characters
  //  2. analyse relation (determine factor.Type)
  //  3. if static relation -> proceed as normal (BUSY)
  //  4. call analysis procedures (as normal)
  //  5. save string to the appropriate file
  if (s == "NewRelations!")
    {
      daten.set_TCP_NODELAY(); // for faster communication
      string ClientAccount = client.substr(0,client.find_first_of(' ')); // default account name for statistical accounting

      {
        // verify, that relations belong to our factorization
        mpz_t received_kN;
        mpz_init(received_kN);
        daten >> received_kN;
        if (daten.fail() || mpz_cmp(received_kN,kN)!=0)
         {
           daten.clear(daten.rdstate() & ~std::ios::failbit);
           daten << "VOID_number!!" << endl;
           return;
         }
        else daten << "proceed" << endl;
        mpz_clear(received_kN);

        daten >> s;
        if (s=="Account:")
         {
           // clients want to register with an alias account name
           s=read_restrictedName(daten); // needed for security reasons to prevent escape sequences or similar bad things!
           if (s!="")
            {
              ClientAccount=s; // okay, use this alias
#ifdef VERBOSE_NOTICE
              cout << client << " registered as \"" << ClientAccount << "\"" << endl;
#endif
            }
           daten >> s; // and now we expect to read keyword for getting a relation
         }
        if (s!="RL") // new token for "Relation!"
         {
           cerr << "expected block of relations, but got nonsense!" << endl;
           return;
         }
      }

      bool Relationsblock_okay = true;
      int received_relations_counter = 0;
      
      statistical_data::AllClientStats_Mutex.lock();
      statistical_data::CClientStats &ClientStats
       = statistical_data::AllClientStats[ClientAccount];
      statistical_data::AllClientStats_Mutex.unlock();
      // generate (or access) and reference a statistical "worksheet" for
      // the specified client (which is given by a string)

      /* Was passiert hier? (auf deutsch)

         "AllClientStats" ist eine Map, d.h. sie verhält sich wie ein Array, auf das
         mit beliebigen Indices eines vorgegebenen Typs zugegriffen werden kann. In unserem
         Fall ist der Index von Typ "String". Wenn unter dem Index ein Eintrag gefunden wurde,
         so wird eine Referenz auf diesen zurückgeliefert. Wenn kein Eintrag gefunden wurde,
         so wird automatisch ein neuer Eintrag angelegt und ebenfalls eine Referenz davon
         zurückgegeben. -- Diesen Eintrag machen wir nun für den Rest dieser Funktion unter der
         handlicheren Referenz "&ClientStats" zugänglich.

         Da die Ermittlung des Eintrags ein kritischer Abschnitt ist (nicht threadsafe!), muß er
         durch Mutex vor konkurrierenden Zugriffen anderer Threads geschützt werden.
         Der Zugriff auf "&ClientStats" ist ebenfalls (innerhalb der Methoden der Klasse CClientStats)
         vor konkurrierenden Zugriffen geschützt.

         Zugriffsbeispiele:

          Szenario 1:
              Ein client mit dem Verbindungsnamen <ip> <port> "127.0.0.1 1000"
              hat sich erstmals an den Server verbunden.
           -> Beim Zugriff auf AllClientStats["127.0.0.1"] wird automatisch
              eine neue Instanz vom Typ CClientStats erzeugt.
              Diese wird für den weiteren Gebrauch innerhalb dieser Funktion
              unter "&ClientStats" verfügbar gemacht.

          Szenario 2:
              Ein anderer client mit dem Verbindungsnamen "myname.test.org 9181" hat sich
              bereits zum zweiten Mal connected.
           -> Der Zugriff findet über AllClientStats["myname.test.org"] statt
              und die bereits existierende Instanz wird für den weiteren
              Gebrauch innerhalb dieser Funktion unter "&ClientStats"
              verfügbar gemacht.
      */

      // additional remark:
      // concurrent access is sequenced by a MUTEX inside CClientStats!

      ClientStats.PutTimeStamp(); // put time stamp on for this connection

      do // receive a block of relations
        {
          CriticalSection.leave();
          // allow other threads to sneak in, because this
          // loop can take really a long time if we would do it atomically!

          //cout << "receiving relation from client." << endl;

          ostringstream os; // read relation into this string-stream for further processing

          //cout << "scanning relation" << endl;
          daten >> s; // starting symbol "G"
          if (s != "G")
           {
             MARK;
             cerr << "error: relation start symbol 'G' expected!" << endl;
#ifdef SAFEMODE
             return; // abort this connection...
#else
             exit(1); // since we are in an trusted environment, something weired must have happened!
#endif
           }

          CmpqsFactor factor;
          daten >> factor; // read factor (dynamic-factor or special-factor of relation or 1 for static relation)
          os << "G " << setprecision(20) << factor; // << " ";
          while (daten.peek()!='\n')
           {
             char line[1024];
             daten.get(line,sizeof(line),'\n'); os << line;
             if (daten.fail())
              {
                cerr << "stream in failstate for " << client << ", aborting thread." << endl;
                return;
              } 
           }
          { char c; daten.get(c); os << c; }
          //cout << os.str();
      
          //daten.ignore(1,'\n');
          daten >> s;

          while (s=="Relationsblock_Sync")
           {
             // client wishes to sync with server
#ifdef VERBOSE_INFO
             cout << "syncing with " << client << ", " << received_relations_counter << " received so far." << endl;
#endif
             daten << "synced. " << flush;
             //cout << "synced with " << client << endl;
             daten >> s;
             if (daten.fail()) return;
             ClientStats.PutTimeStamp(); // put time stamp on for this connection
             CriticalSection.enter(); StatusReport(); CriticalSection.leave();
           }

          received_relations_counter++;

#if 0
          // this version is less verbose
          CriticalSection.enter(); // still each step should be processed atomically...
          // okay, now we are inside the critical section
#else
          // this version is more verbose; use it to detect lifelocks

          // [
          //   Assume you are at the crossing of two streets, then a
          //   DEADLOCK is a situation, where at each of the four
          //   entranceways there is a car waiting to enter the crossing.
          //
          //   A LIFELOCK is a different situation, where you want to cross
          //   a street, but there is so much traffic, that you cannot
          //   actually do it.
          //   
          //   Now assume a LIFELOCK, which is caused by people who are
          //   searching for a parking place. But you cannot leave your
          //   parking place because of the traffic generated by this
          //   LIFELOCK. Actually this can lead to a situation, that is very
          //   similar to a DEADLOCK, because the traffic gets jammed.
          //
          //   Well, DEADLOCKS and LIFELOCKS can be avoided by design in
          //   multithreaded programs. But you might still be interested to
          //   measure the traffic for detecting design flaws...
          // ]

          // still each step should be processed atomically...
          if (!CriticalSection.try_enter())
           {
             clock_t Start = times(NULL);
             CriticalSection.enter();
             clock_t Stop = times(NULL); double w=difftimes(Stop,Start);
             if (w>0.04) // print any client that got suspended for more than 0.04 seconds
               cout << "thread " <<  pthread_self() << ", " << client
                    << " suspended for " << difftimes(Stop,Start) << " seconds." << endl;
           }
          // okay, critical section entered
#endif


#ifdef SAFEMODE
          {
            // now check, whether received relation is valid
            istringstream is(os.str()); // provide read string as input
            if (!CRelation::is_valid(is))
             {
            #ifdef VERBOSE_WARN
               MARK;
               cerr << "invalid relation received from " << client << endl;
            #endif
               Relationsblock_okay=false;
               continue; // or should we break?
             }
            else
             {
            #ifdef VERBOSE_INFO
               cout << "valid relation received from " << client << endl;
            #endif
             }
          }
#endif
          
          ClientStats.increment(factor.Type()); // increment statistical counter for this type of factor
          if (factor.IsTypeOf(CmpqsFactor::static_prime))
           {
             // Static Factor
             //cout << "Received static relation!" << endl;
             statistical_data::DynamicFactorRating.increment_Hits_at_position(0); // 0 dynamic factors directly involved (serversided)
             CriticalSection.leave();

             istringstream is(os.str()); // provide read string as input
             CRelation* GL = new CRelation();
             GL->set_MulticombineData(new CRelation::SMulticombineData);
             GL->multi_combine_init();
             factor = GL->multi_combine_main(is);
             //cout << "inserting relation into system of equations." << endl;
             StaticRelations::insert(GL,without_multi_combine_init);
             continue;
           }
          if (factor.IsTypeOf(CmpqsFactor::single_large_prime))
           {
            // Dynamic Factor
            TDynamicFactorRelation relation;
            relation.factor = factor.int_value();
            if (!is_dynamic_factor(relation)) // known dynamic factor? -> if so, then fill in the missing data 
              { // a new single large prime!
                relation.fpos=DynamicRelations_to_file.tellp();
                DynamicRelations_to_file << os.str(); // fast save, because the string is written without further conversion
                  // IMPORTANT:
                  // Normally we need to flush DynamicRelations_to_file here to ensure, that
                  // any istream pointing to the same file is up-to-date (because of possible
                  // correction factors ["Korrekturfaktoren"]). But this would slow down performance!
                  // (Flushing is only relevant, iff unflushed correction factors occur in static relations,
                  // and compared to the number of incoming SLP relations such events are really seldom...)
                  // To ensure integrity, we do the following: For the combine routines, we catch failing
                  // seeks and do error handling... 
                // relation.append_for_sieving(); // server need not to do this!
                DynamicFactorRelations.insert(relation);
                SpecialRelations::split_by_primefactor(relation.factor);
              } 
            else
              { // factor already known!
                //cout << "Can convert dynamic relation to a static relation!" << endl;
                statistical_data::DynamicFactorRating.increment_Hits_at_position(1);
                CriticalSection.leave();

                istringstream is(os.str()); // provide read string as input
                CRelation* GL = new CRelation();
                GL->set_MulticombineData(new CRelation::SMulticombineData);
                GL->multi_combine_init();
                factor = GL->multi_combine_main(is);
                {
                  mpz_t x; mpz_init(x);
                  factor.assign_to_mpz(x); // store factor into mpz-variable x
                  mpz_mul(GL->Delta,GL->Delta,x); mpz_mod(GL->Delta,GL->Delta,n);
                  mpz_clear(x);
                }
                istream *pis = DynamicRelations::IstreamPool::acquire_istream();
                GL->multi_combine_main(*pis, relation.fpos);
                DynamicRelations::IstreamPool::release_istream(pis);
                //GL->multi_combine_exit();
                StaticRelations::insert(GL,without_multi_combine_init);
              }
            continue;
           }
          if (factor.IsTypeOf(CmpqsFactor::double_large_prime))
           {
             // Special-Factor

             // Specialfactor mit Dynamicfactoren testen:
             /* Anmerkung:
                 Beim verteilten Sieben kann es passieren, dass ein (notwendigerweise zusammengesetzter)
                 Special-Factor von einem Client entdeckt wird, der nicht mit dem aktuellen Satz der
                 dynamischen Faktoren siebt. - In diesem Fall, der mit steigender Anzahl der Clients immer
                 wahrscheinlicher wird, könnte es sinnvoll sein, Specialfaktoren auf ihre Teilbarkeit
                 mit den vorliegenden dynamischen Faktoren des Servers zu testen und ggf. zu splitten.
                 Allerdings bringt ein Verzicht auf diesen Test keine weiteren Nachteile mit sich.
                 Wir haben daher auf diese - nur beim verteilten Sieben mögliche - Prozedur bisher verzichtet.

                 Nachtrag 2004-05-02:
                 Ist jetzt wieder aktuell geworden, da ich eine Option eingebaut habe,
                 nur bis zu einer Obergrenze der dynamischen Faktoren zu sieben. Der Rest der dynamischen
                 Faktoren verhält sich also passiv und könnte durchaus im Special-Factor enthalten sein,
                 wenn der Client ihn nicht bereits entfernt und die Relation als Single-Large-Prime-Relation
                 klassifiziert hat. Und das kann durchaus passieren, wenn der Client nicht im Besitz der vollen
                 Menge der SingleLargePrimes ist!
                 Wir machen es jetzt auf die unelegante Tour mit einer vollen Breitseite:
                  -> zuerst wird die Relation als Special-Relation aufgenommen
                  -> Dann werden ihre beiden Teilfaktoren auf mögliche SingleLargePrimes überprüft,
                     und (sollte das der Fall sein):
                      -> SpecialRelations::split_by_primefactor() aufgerufen.
              */
             SpecialRelations::insert(factor,os.str());
             if (is_dynamic_factor(factor.LP1())) SpecialRelations::split_by_primefactor(factor.LP1());
             if (is_dynamic_factor(factor.LP2())) SpecialRelations::split_by_primefactor(factor.LP2());
             continue;
           }

         cerr << "Error: Trash or a relation of unknown type has been received!" << endl;
         Relationsblock_okay=false;
        } while (s=="RL"); // new token for "Relation!"
      // block of relations has been read
      CriticalSection.leave();
      cout_network << "Block of " << received_relations_counter 
                   << " relations received from " << client << "." << endl;
      if (!Relationsblock_okay) daten << "ignoriert!" << flush;
      else
       if (s=="Relationsblock_Ende") daten << "empfangen." << flush;
       else
        {
          cerr << "could not detect (expected) end of block of relations!" << endl;
          daten << "Relationsblock_korrekt_beendet???" << flush;
        }
      CriticalSection.enter(); StatusReport(); CriticalSection.leave();
      return;
    }

  cerr << "Invalid request \"" << s << "\""
       << " from " << client << endl;
}


void* Cprocess_clients::THREAD_process_data_stream(void* arg)
{
  ExitManager::register_cancel();
  const int communication_socket = *static_cast<int*>(arg);
  delete static_cast<int*>(arg);

  unix_io_stream *communication_stream = NULL;

  try
   {
     communication_stream = new unix_io_stream(communication_socket);
   }
  catch (exception &e)
   {
     cerr << "while constructing caught an exception: " << e.what() << endl;
     goto done;
   }
  catch (...)
   {
     cerr << "while constructing caught an unknown exception!" << endl;
     throw;
   }

  //cout << "processing thread..." << endl;

  try
   {
     process_data_stream(*communication_stream,peer_info(*communication_stream));
   }
  catch (exception &e)
   {
     cerr << "Cprocess_clients::THREAD_process_data_stream: caught an exception: " << e.what() << endl;
     goto done;
   }
  catch (...)
   {
     cerr << "Cprocess_clients::THREAD_process_data_stream: while trying to process data stream caught an unknown exception!" << endl;
     throw;
   }

  //cout << "... thread processed!" << endl;

  try
   {
     if (!communication_stream->fail()) communication_stream->flush();
   }
  catch (exception &e)
   {
     cerr << "caught an exception: " << e.what() << endl;
   }
  catch (...)
   {
     cerr << "caught an unknown exception!" << endl;
     throw;
   }

done:

  try
   {
     delete communication_stream;
   }
  catch (exception &e)
   {
     cerr << "while destructing caught an exception: " << e.what() << endl;
   }
  catch (...)
   {
     cerr << "while destructing caught an unknown exception!" << endl;
     throw;
   }

  ExitManager::unregister_cancel();
  return 0;
}

---