ToolDirectoryServer.C

Go to the documentation of this file.

// Copyright (C) 2001, Compaq Computer Corporation
// 
// This file is part of Vesta.
// 
// Vesta is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
// 
// Vesta is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
// 
// You should have received a copy of the GNU Lesser General Public
// License along with Vesta; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

// File: ToolDirectoryServer.C
// Last modified on Mon May 23 23:26:00 EDT 2005 by ken@xorian.net         
//      modified on Tue Apr 26 09:46:01 EDT 2005 by irina.furman@intel.com 
//      modified on Sat Jun  2 14:21:01 PDT 2001 by mann   
//      modified on Fri Jun  1 12:22:20 PDT 2001 by yuanyu 
//      modified on Mon Apr 21 16:32:23 PDT 1997 by heydon 
//      modified on Fri Mar  1 12:38:47 PST 1996 by levin  
//      modified on Fri Jan 26 19:14:25 PST 1996 by horning

#include <Basics.H>
#include <Atom.H>
#include <SRPC.H>
#include <LimService.H>
#include <VestaConfig.H>
#include "ToolDirectoryServer.H"
#include "EvalBasics.H"
#include "Val.H"
#include "Expr.H"
#include "Evaluator_Dir_SRPC.H"

#include <sstream>

using std::ostringstream;
using std::cin;
using std::cerr;
using std::endl;
using std::flush;

static Basics::mutex TDSMu;

static Text tmpName("tmp");

// size of ToolDirectoryServer thread stacks (in bytes).
const long CallbackStackSize = 0x20000;

// Used to record runtool volatile directories and dependency:
RunToolCalls runToolCalls(maxThreads);

static pthread_once_t init_block = PTHREAD_ONCE_INIT;

// Tool directory server request counter.
static Bit32 toolDirServerReqs = 0UL;

// Tool call counter.
static Bit64 toolCalls = 0UL;

Word GetNewHandle() {
  TDSMu.lock();
  toolDirServerReqs++;
  toolCalls++;
  Word newHandle = ((toolCalls<<20) |
                    (toolDirServerReqs & 0xFFFFF));
  TDSMu.unlock();
  return newHandle;
}

Word GetNewHandle(Word handle) {
  TDSMu.lock();
  toolDirServerReqs++;
  Word newHandle = ((handle & CONST_INT_64(0xFFFFFFFFFFF00000)) |
                    (toolDirServerReqs & 0xFFFFF));
  TDSMu.unlock();
  return newHandle;
}

// Marshal/unmarshal directory handles.
void send_dir(SRPC *srpc, Word dirHandle) {
  DirInfos *dirInfos;
  assert(runToolCalls.Get(WordKey(dirHandle>>20), dirInfos));
  srpc->send_bytes((char*)&dirHandle, DIR_HANDLE_BYTES);
}

DirInfo *recv_dir(SRPC *srpc, /*OUT*/ Word& dirHandle, /*OUT*/ DirInfos*& dirInfos) {
  int len = DIR_HANDLE_BYTES;
  srpc->recv_bytes_here((char*)&dirHandle, len);
  if (len != DIR_HANDLE_BYTES) {
    throw(SRPC::failure(SRPC::invalid_parameter, "Invalid directory handle"));
  }
  if (!runToolCalls.Get(WordKey(dirHandle>>20), dirInfos)) {
    throw(SRPC::failure(SRPC::invalid_parameter, "No runtool for this directory handle"));
  }
  DirInfo *dirInfo;
  dirInfos->mu.lock();
  bool inTbl = dirInfos->dirInfoTbl->Get(WordKey(dirHandle), dirInfo);
  dirInfos->mu.unlock();
  if (!inTbl) {
    throw(SRPC::failure(SRPC::invalid_parameter, "No such directory handle"));
  }
  return dirInfo;
}

void DirInfos::AddDpnd(DirInfo *dir, const Text& name, Val v, const PathKind pk) {
  if (dir->coarseDep) return;

  dir->child_lookup = true;
  
  DepPath newDepPath(dir->path.content);
  if (pk == BLenPK) {
    newDepPath.content->pKind = BLenPK;
    v = ((BindingVC*)v)->Names();
  }
  else {
    newDepPath.Extend(name, pk);
  }
  if (fsDeps) {
    outputMu.lock();
    cerr << "FS dependency: ";
    newDepPath.Print(&cerr);
    cerr << "\n";
    outputMu.unlock();    
  }
  DepPathTbl::KVPairPtr pr;
  this->mu.lock();
  this->dep->Put(newDepPath, v, pr);
  this->mu.unlock();
  // assert(v->FingerPrint() == pr->val->FingerPrint());
}

Word DirInfos::LookupDir(DirInfo *dir, const Text& name, Val v, Word dirHandle) {
  /* Assumption: the repository does not look up the same directory
     more than once during a runtool call.  We do not keep a record
     of the directories being visited, and create a new dir handle
     for every directory lookup request.   */
  DepPath newPath(dir->path.content);

  newPath.Extend(name);
  DirInfo *newDirInfo = NEW(DirInfo);
  newDirInfo->b = (BindingVC*)v;
  newDirInfo->path = newPath;
  newDirInfo->isRoot = false;
  newDirInfo->coarseDep = (dir->isRoot && name == tmpName);

  newDirInfo->child_lookup = false;
  dir->child_lookup = true;

  this->mu.lock();

  if (newDirInfo->coarseDep) {
    // Make it depend on the entire directory content.
    DepPath newDepPath(newDirInfo->path.content);
    if (fsDeps) {
      outputMu.lock();
      cerr << "FS dependency: ";
      newDepPath.Print(&cerr);
      cerr << "\n";
      outputMu.unlock();    
    }
    DepPathTbl::KVPairPtr pr;
    this->dep->Put(newDepPath, v, pr);
  }
  
  // Get a new handle of this new directory:
  Word newDirHandle = GetNewHandle(dirHandle);
  this->dirInfoTbl->Put(WordKey(newDirHandle), newDirInfo);

  this->mu.unlock();

  return newDirHandle;
}

// Server procedure:
void Evaluator_Dir_Server_Inner(SRPC *srpc, int intfVersion, int procId, void *arg)
  throw(SRPC::failure, Evaluator::failure)
{
  int index;
  Val v;
  DirInfo *dir;
  Word dirHandle;
  DirInfos *dirs;

  if (intfVersion != EVALUATOR_DIR_SRPC_VERSION) {
    srpc->send_failure(SRPC::version_skew, "Unrecognized intfVersion");
  }

  // Dispatch procedure:
  switch(procId) {
  case ed_lookup:
    {
      dir = recv_dir(srpc, dirHandle, dirs);
      // recv_chars() allocates the bytes on the heap. So, we do not
      // need to copy the bytes for id.
      Atom id(srpc->recv_chars(), (void*)1);
      srpc->recv_end();
      if (evalCalls) {
        outputMu.lock();        
        cerr << "EvalDir lookup: ";
        dir->path.Print(&cerr);
        cerr << '/' << id << '\n';
        outputMu.unlock();
      }
      v = dir->b->GetElemNoDpnd(id, index);
      if (v->vKind != ErrorVK) {
        switch (v->vKind) {
        case BindingVK:
          srpc->send_int(ed_directory);
          srpc->send_int(index);
          send_dir(srpc, dirs->LookupDir(dir, id, v, dirHandle));
          break;
        case TextVK:
          dirs->AddDpnd(dir, id, v, NormPK);
          srpc->send_int(ed_file);
          srpc->send_int(index);
          srpc->send_int(((TextVC*)v)->Sid());
          ((TextVC*)v)->FingerPrint().Send(*srpc);
          break;
        case ModelVK:
          dirs->AddDpnd(dir, id, v, NormPK);
          srpc->send_int(ed_file);
          srpc->send_int(index);
          srpc->send_int(((ModelVC*)v)->Sid());
          ((ModelVC*)v)->FingerPrintFile().Send(*srpc);
          break;
        case IntegerVK:
          dirs->AddDpnd(dir, id, v, NormPK);
          srpc->send_int(ed_device);
          srpc->send_int(index);
          srpc->send_int(((IntegerVC*)v)->num);
          break;
        default:
          throw(SRPC::failure(SRPC::invalid_parameter,
                              id + " is not a file or directory"));
        }
      }
      else {
        dirs->AddDpnd(dir, id, valFalse, BangPK);
        srpc->send_int(ed_none);
      }
      srpc->send_end();
      break;
    }
  case ed_lookup_index:
    {
      Text id;
      dir = recv_dir(srpc, dirHandle, dirs);
      index = srpc->recv_int();
      srpc->recv_end();
      if (evalCalls) {
        outputMu.lock();        
        cerr << "EvalDir index:  ";
        dir->path.Print(&cerr);
        cerr << '/' << index << '\n';
        outputMu.unlock();              
      }
      v = dir->b->GetElemNoDpnd(index, id);
      if (v->vKind != ErrorVK) {
        switch (v->vKind) {
        case BindingVK:
          srpc->send_int(ed_directory);
          srpc->send_chars(id.chars());
          send_dir(srpc, dirs->LookupDir(dir, id, v, dirHandle));
          break;
        case TextVK:
          dirs->AddDpnd(dir, id, v, NormPK);
          srpc->send_int(ed_file);
          srpc->send_chars(id.chars());
          srpc->send_int(((TextVC*)v)->Sid());
          ((TextVC*)v)->FingerPrint().Send(*srpc);
          break;
        case ModelVK:
          dirs->AddDpnd(dir, id, v, NormPK);
          srpc->send_int(ed_file);
          srpc->send_chars(id.chars());
          srpc->send_int(((ModelVC*)v)->Sid());
          ((ModelVC*)v)->FingerPrintFile().Send(*srpc);
          break;
        case IntegerVK:
          dirs->AddDpnd(dir, id, v, NormPK);
          srpc->send_int(ed_device);
          srpc->send_chars(id.chars());
          srpc->send_int(((IntegerVC*)v)->num);
          break;
        default:
          throw(SRPC::failure(SRPC::invalid_parameter,
                              id + " is not a file or directory"));
        }
      }
      else {
        dirs->AddDpnd(dir, id, valFalse, BangPK);
        srpc->send_int(ed_none);
      }
      srpc->send_end();
      break;
    }
  case ed_oldlist:
  case ed_list:
    {
      Text id;
      dir = recv_dir(srpc, dirHandle, dirs);
      index = srpc->recv_int();
      int limit = srpc->recv_int();
      int overhead = srpc->recv_int();
      srpc->recv_end();
      if (evalCalls) {
        outputMu.lock();                
        cerr << "EvalDir list:   ";
        dir->path.Print(&cerr);
        cerr << '/' << index << '\n';
        outputMu.unlock();              
      }
      dirs->AddDpnd(dir, emptyText, dir->b, BLenPK);
      srpc->send_seq_start();
      while (true) {
        v = dir->b->GetElemNoDpnd(index, id);
        if (v->vKind == ErrorVK) {
          // Send terminating entry.
          srpc->send_chars("");
          srpc->send_int(ed_none);
          if (procId != ed_oldlist) {
            srpc->send_int(NullShortId);
          }
          break;
        }
        // Directory entry found; check limit
        limit -= id.Length()+overhead;
        if (limit < 0) break;
        // Send this entry
        srpc->send_chars(id.chars());
        ed_entry_type etype =
          (v->vKind == TextVK) ? ed_file :
          (v->vKind == BindingVK) ? ed_directory :
          (v->vKind == IntegerVK) ? ed_device : ed_none;
        if (etype == ed_none) 
          throw(SRPC::failure(SRPC::invalid_parameter,
                              id + " is not a file or directory"));
        srpc->send_int(etype);
        if (procId != ed_oldlist) {
          if (etype == ed_file) {
            srpc->send_int(((TextVC*)v)->Sid());
          } else {
            srpc->send_int(NullShortId);
          }
        }
        index++;
      } // while(true)
      srpc->send_seq_end();
      srpc->send_end();
      break;
    }
  default:
    srpc->send_failure(SRPC::invalid_parameter, "Unrecognized procId");
    break;
  }  // switch(procId)
}

void Evaluator_Dir_Server(SRPC *srpc, int intfVersion, int procId, void *arg)
  throw(SRPC::failure)
  /* This is the server for the Evaluator_Dir_SRPC interface. It 
     expects to be invoked via the LimService interface, which
     establishes the RPC connection with the client. 'arg' is unused. */
{
  try {
    Evaluator_Dir_Server_Inner(srpc, intfVersion, procId, arg);
  }
  catch (Evaluator::failure f) {
    outputMu.lock();
    cerr << "Evaluator failure occured in tool directory lookup.\n";
    outputMu.unlock();
    exit(1);
  }
}

Text convert_failure(const SRPC::failure& f) {
  ostringstream s;
  s << "SRPC failure in ToolDirServer (";
  switch (f.r) {
  case SRPC::unknown_host:
    s << "unknown_host"; break;
  case SRPC::unknown_interface:
    s << "unknown_interface"; break;
  case SRPC::version_skew:
    s << "version_skew"; break;
  case SRPC::protocol_violation:
    s << "protocol_violation"; break;
  case SRPC::buffer_too_small:
    s << "buffer_too_small"; break;
  case SRPC::transport_failure:
    s << "transport_failure"; break;
  case SRPC::internal_trouble:
    s << "internal_trouble"; break;
  case SRPC::invalid_parameter:
    s << "invalid_parameter"; break;
  case SRPC::partner_went_away:
    s << "partner_went_away"; break;
  case SRPC::not_implemented:
    s << "not_implemented"; break;
  default:
    s << f.r; break;
  }
  s << "): " << f.msg.chars();
  Text t(s.str());
  return t;
}

void Evaluator_Dir_Server_Failure(SRPC *srpc, const SRPC::failure& f,
                                  void *arg) {
  // called when a failure occurs during server execution
  if (f.r == SRPC::partner_went_away) return; // normal occurrence
  outputMu.lock();        
  cerr << convert_failure(f).chars() << " (continuing...)\n";
  outputMu.unlock();
}

extern "C"
{
  void init() {
    toolCalls = ((Bit64) time(NULL))<<20;
    try {
      // Get the maximum number of simultaneously serviced callbacks
      // from the repository
      int max_running = VestaConfig::get_int("Evaluator", "server_max_threads");

      // Get or determine the maximum number of incoming callbacks from
      // the repository before the evaluator gives a "server busy" error.
      Text max_pending_t;
      int max_pending;
      if(!VestaConfig::get("Evaluator", "server_max_pending", max_pending_t) ||
         (max_pending_t == "auto"))
        {
          // The maximum number of simultaneous requests we expect.
          // (Probably [Repository]threads alone is sufficient, but if
          // the number of evaluator threads is higher, we'll use that.)
          int max_needed = max(maxThreads,
                               VestaConfig::get_int("Repository", "threads"));

          max_pending = max_needed - max_running;
        }
      else
        {
          // If server_max_pending is set and not "auto", it must be an
          // integer.
          max_pending = VestaConfig::get_int("Evaluator", "server_max_pending");
        }

      LimService *ls =
        NEW_CONSTR(LimService, (max_running, max_pending,
                       Evaluator_Dir_Server, Evaluator_Dir_Server_Failure,
                       /*arg=*/ NULL, CallbackStackSize));
      toolDirServerIntfName = ls->IntfName();
      if (getenv("DEBUG_TOOL_DIR_SERVER")) {
        outputMu.lock();
        cerr << "Tool directory server listening on port "
             << toolDirServerIntfName << endl
             << "Hit enter to continue: " << flush;
        char enter;
        cin.get(enter);
        outputMu.unlock();
      }
      (void) ls->Forked_Run();
    } catch (VestaConfig::failure f) {
      outputMu.lock();    
      cerr << endl
           << "Configuration error starting tool directory server:" << endl
           << f.msg << endl;
      exit(1);
    } catch (SRPC::failure f) {
      outputMu.lock();        
      cerr << endl
           << "Error starting tool directory server:" << endl
           << convert_failure(f) << endl;
      outputMu.unlock();
      exit(1);
    } catch(...) {
      outputMu.lock();        
      cerr << endl
           << "Unexpected exception starting tool directory server!" << endl;
      outputMu.unlock();
      exit(1);
    }
  }
}

void create_tool_directory_server() {
  pthread_once(&init_block, &init);
}

---