apt-pkg/acquire.cc

   1 // -*- mode: cpp; mode: fold -*-
   2 // Description                                                          /*{{{*/
   3 // $Id: acquire.cc,v 1.8 1998/11/05 07:21:40 jgg Exp $
   4 /* ######################################################################
   5
   6    Acquire - File Acquiration
   7
   8    The core element for the schedual system is the concept of a named
   9    queue. Each queue is unique and each queue has a name derived from the
  10    URI. The degree of paralization can be controled by how the queue
  11    name is derived from the URI.
  12
  13    ##################################################################### */
  14                                                                         /*}}}*/
  15 // Include Files                                                        /*{{{*/
  16 #ifdef __GNUG__
  17 #pragma implementation "apt-pkg/acquire.h"
  18 #endif
  19 #include <apt-pkg/acquire.h>
  20 #include <apt-pkg/acquire-item.h>
  21 #include <apt-pkg/acquire-worker.h>
  22 #include <apt-pkg/configuration.h>
  23 #include <apt-pkg/error.h>
  24 #include <strutl.h>
  25                                                                         /*}}}*/
  26
  27 // Acquire::pkgAcquire - Constructor                                    /*{{{*/
  28 // ---------------------------------------------------------------------
  29 /* We grab some runtime state from the configuration space */
  30 pkgAcquire::pkgAcquire()
  31 {
  32    Queues = 0;
  33    Configs = 0;
  34    Workers = 0;
  35    ToFetch = 0;
  36    Running = false;
  37
  38    string Mode = _config->Find("Acquire::Queue-Mode","host");
  39    if (strcasecmp(Mode.c_str(),"host") == 0)
  40       QueueMode = QueueHost;
  41    if (strcasecmp(Mode.c_str(),"access") == 0)
  42       QueueMode = QueueAccess;
  43
  44    Debug = _config->FindB("Debug::pkgAcquire",false);
  45 }
  46                                                                         /*}}}*/
  47 // Acquire::~pkgAcquire - Destructor                                    /*{{{*/
  48 // ---------------------------------------------------------------------
  49 /* Free our memory, clean up the queues (destroy the workers) */
  50 pkgAcquire::~pkgAcquire()
  51 {
  52    while (Items.size() != 0)
  53       delete Items[0];
  54
  55    while (Configs != 0)
  56    {
  57       MethodConfig *Jnk = Configs;
  58       Configs = Configs->Next;
  59       delete Jnk;
  60    }
  61
  62    while (Queues != 0)
  63    {
  64       Queue *Jnk = Queues;
  65       Queues = Queues->Next;
  66       delete Jnk;
  67    }
  68 }
  69                                                                         /*}}}*/
  70 // Acquire::Add - Add a new item                                        /*{{{*/
  71 // ---------------------------------------------------------------------
  72 /* This puts an item on the acquire list. This list is mainly for tracking
  73    item status */
  74 void pkgAcquire::Add(Item *Itm)
  75 {
  76    Items.push_back(Itm);
  77 }
  78                                                                         /*}}}*/
  79 // Acquire::Remove - Remove a item                                      /*{{{*/
  80 // ---------------------------------------------------------------------
  81 /* Remove an item from the acquire list. This is usually not used.. */
  82 void pkgAcquire::Remove(Item *Itm)
  83 {
  84    for (vector<Item *>::iterator I = Items.begin(); I < Items.end(); I++)
  85    {
  86       if (*I == Itm)
  87          Items.erase(I);
  88    }
  89 }
  90                                                                         /*}}}*/
  91 // Acquire::Add - Add a worker                                          /*{{{*/
  92 // ---------------------------------------------------------------------
  93 /* A list of workers is kept so that the select loop can direct their FD
  94    usage. */
  95 void pkgAcquire::Add(Worker *Work)
  96 {
  97    Work->NextAcquire = Workers;
  98    Workers = Work;
  99 }
 100                                                                         /*}}}*/
 101 // Acquire::Remove - Remove a worker                                    /*{{{*/
 102 // ---------------------------------------------------------------------
 103 /* A worker has died. This can not be done while the select loop is running
 104    as it would require that RunFds could handling a changing list state and
 105    it cant.. */
 106 void pkgAcquire::Remove(Worker *Work)
 107 {
 108    if (Running == true)
 109       abort();
 110
 111    Worker **I = &Workers;
 112    for (; *I != 0;)
 113    {
 114       if (*I == Work)
 115          *I = (*I)->NextAcquire;
 116       else
 117          I = &(*I)->NextAcquire;
 118    }
 119 }
 120                                                                         /*}}}*/
 121 // Acquire::Enqueue - Queue an URI for fetching                         /*{{{*/
 122 // ---------------------------------------------------------------------
 123 /* This is the entry point for an item. An item calls this function when
 124    it is construction which creates a queue (based on the current queue
 125    mode) and puts the item in that queue. If the system is running then
 126    the queue might be started. */
 127 void pkgAcquire::Enqueue(Item *Itm,string URI,string Description)
 128 {
 129    // Determine which queue to put the item in
 130    string Name = QueueName(URI);
 131    if (Name.empty() == true)
 132       return;
 133
 134    // Find the queue structure
 135    Queue *I = Queues;
 136    for (; I != 0 && I->Name != Name; I = I->Next);
 137    if (I == 0)
 138    {
 139       I = new Queue(Name,this);
 140       I->Next = Queues;
 141       Queues = I;
 142
 143       if (Running == true)
 144          I->Startup();
 145    }
 146
 147    Itm->Status = Item::StatIdle;
 148
 149    // Queue it into the named queue
 150    I->Enqueue(Itm,URI,Description);
 151    ToFetch++;
 152
 153    // Some trace stuff
 154    if (Debug == true)
 155    {
 156       clog << "Fetching " << URI << endl;
 157       clog << " to " << Itm->DestFile << endl;
 158       clog << " Queue is: " << QueueName(URI) << endl;
 159    }
 160 }
 161                                                                         /*}}}*/
 162 // Acquire::Dequeue - Remove an item from all queues                    /*{{{*/
 163 // ---------------------------------------------------------------------
 164 /* This is called when an item is finished being fetched. It removes it
 165    from all the queues */
 166 void pkgAcquire::Dequeue(Item *Itm)
 167 {
 168    Queue *I = Queues;
 169    bool Res = false;
 170    for (; I != 0; I = I->Next)
 171       Res |= I->Dequeue(Itm);
 172
 173    if (Debug == true)
 174       clog << "Dequeuing " << Itm->DestFile << endl;
 175    if (Res == true)
 176       ToFetch--;
 177 }
 178                                                                         /*}}}*/
 179 // Acquire::QueueName - Return the name of the queue for this URI       /*{{{*/
 180 // ---------------------------------------------------------------------
 181 /* The string returned depends on the configuration settings and the
 182    method parameters. Given something like http://foo.org/bar it can
 183    return http://foo.org or http */
 184 string pkgAcquire::QueueName(string Uri)
 185 {
 186    URI U(Uri);
 187
 188    const MethodConfig *Config = GetConfig(U.Access);
 189    if (Config == 0)
 190       return string();
 191
 192    /* Single-Instance methods get exactly one queue per URI. This is
 193       also used for the Access queue method  */
 194    if (Config->SingleInstance == true || QueueMode == QueueAccess)
 195       return U.Access;
 196
 197    return U.Access + ':' + U.Host;
 198 }
 199                                                                         /*}}}*/
 200 // Acquire::GetConfig - Fetch the configuration information             /*{{{*/
 201 // ---------------------------------------------------------------------
 202 /* This locates the configuration structure for an access method. If
 203    a config structure cannot be found a Worker will be created to
 204    retrieve it */
 205 pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
 206 {
 207    // Search for an existing config
 208    MethodConfig *Conf;
 209    for (Conf = Configs; Conf != 0; Conf = Conf->Next)
 210       if (Conf->Access == Access)
 211          return Conf;
 212
 213    // Create the new config class
 214    Conf = new MethodConfig;
 215    Conf->Access = Access;
 216    Conf->Next = Configs;
 217    Configs = Conf;
 218
 219    // Create the worker to fetch the configuration
 220    Worker Work(Conf);
 221    if (Work.Start() == false)
 222       return 0;
 223
 224    return Conf;
 225 }
 226                                                                         /*}}}*/
 227 // Acquire::SetFds - Deal with readable FDs                             /*{{{*/
 228 // ---------------------------------------------------------------------
 229 /* Collect FDs that have activity monitors into the fd sets */
 230 void pkgAcquire::SetFds(int &Fd,fd_set *RSet,fd_set *WSet)
 231 {
 232    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 233    {
 234       if (I->InReady == true && I->InFd >= 0)
 235       {
 236          if (Fd < I->InFd)
 237             Fd = I->InFd;
 238          FD_SET(I->InFd,RSet);
 239       }
 240       if (I->OutReady == true && I->OutFd >= 0)
 241       {
 242          if (Fd < I->OutFd)
 243             Fd = I->OutFd;
 244          FD_SET(I->OutFd,WSet);
 245       }
 246    }
 247 }
 248                                                                         /*}}}*/
 249 // Acquire::RunFds - Deal with active FDs                               /*{{{*/
 250 // ---------------------------------------------------------------------
 251 /* Dispatch active FDs over to the proper workers. It is very important
 252    that a worker never be erased while this is running! The queue class
 253    should never erase a worker except during shutdown processing. */
 254 void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet)
 255 {
 256    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 257    {
 258       if (I->InFd >= 0 && FD_ISSET(I->InFd,RSet) != 0)
 259          I->InFdReady();
 260       if (I->OutFd >= 0 && FD_ISSET(I->OutFd,WSet) != 0)
 261          I->OutFdReady();
 262    }
 263 }
 264                                                                         /*}}}*/
 265 // Acquire::Run - Run the fetch sequence                                /*{{{*/
 266 // ---------------------------------------------------------------------
 267 /* This runs the queues. It manages a select loop for all of the
 268    Worker tasks. The workers interact with the queues and items to
 269    manage the actual fetch. */
 270 bool pkgAcquire::Run()
 271 {
 272    Running = true;
 273
 274    for (Queue *I = Queues; I != 0; I = I->Next)
 275       I->Startup();
 276
 277    // Run till all things have been acquired
 278    while (ToFetch > 0)
 279    {
 280       fd_set RFds;
 281       fd_set WFds;
 282       int Highest = 0;
 283       FD_ZERO(&RFds);
 284       FD_ZERO(&WFds);
 285       SetFds(Highest,&RFds,&WFds);
 286
 287       if (select(Highest+1,&RFds,&WFds,0,0) <= 0)
 288       {
 289          Running = false;
 290          return _error->Errno("select","Select has failed");
 291       }
 292
 293       RunFds(&RFds,&WFds);
 294       if (_error->PendingError() == true)
 295          break;
 296    }
 297
 298    // Shut down the acquire bits
 299    Running = false;
 300    for (Queue *I = Queues; I != 0; I = I->Next)
 301       I->Shutdown();
 302
 303    return _error->PendingError();
 304 }
 305                                                                         /*}}}*/
 306 // Acquire::Bump - Called when an item is dequeued                      /*{{{*/
 307 // ---------------------------------------------------------------------
 308 /* This routine bumps idle queues in hopes that they will be able to fetch
 309    the dequeued item */
 310 void pkgAcquire::Bump()
 311 {
 312    for (Queue *I = Queues; I != 0; I = I->Next)
 313       I->Bump();
 314 }
 315                                                                         /*}}}*/
 316
 317 // Acquire::MethodConfig::MethodConfig - Constructor                    /*{{{*/
 318 // ---------------------------------------------------------------------
 319 /* */
 320 pkgAcquire::MethodConfig::MethodConfig()
 321 {
 322    SingleInstance = false;
 323    PreScan = false;
 324    Pipeline = false;
 325    SendConfig = false;
 326    Next = 0;
 327 }
 328                                                                         /*}}}*/
 329
 330 // Queue::Queue - Constructor                                           /*{{{*/
 331 // ---------------------------------------------------------------------
 332 /* */
 333 pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name),
 334             Owner(Owner)
 335 {
 336    Items = 0;
 337    Next = 0;
 338    Workers = 0;
 339 }
 340                                                                         /*}}}*/
 341 // Queue::~Queue - Destructor                                           /*{{{*/
 342 // ---------------------------------------------------------------------
 343 /* */
 344 pkgAcquire::Queue::~Queue()
 345 {
 346    Shutdown();
 347
 348    while (Items != 0)
 349    {
 350       QItem *Jnk = Items;
 351       Items = Items->Next;
 352       delete Jnk;
 353    }
 354 }
 355                                                                         /*}}}*/
 356 // Queue::Enqueue - Queue an item to the queue                          /*{{{*/
 357 // ---------------------------------------------------------------------
 358 /* */
 359 void pkgAcquire::Queue::Enqueue(Item *Owner,string URI,string Description)
 360 {
 361    // Create a new item
 362    QItem *I = new QItem;
 363    I->Next = Items;
 364    Items = I;
 365
 366    // Fill it in
 367    Items->Owner = Owner;
 368    Items->URI = URI;
 369    Items->Description = Description;
 370    Owner->QueueCounter++;
 371
 372    if (Items->Next == 0)
 373       Cycle();
 374 }
 375                                                                         /*}}}*/
 376 // Queue::Dequeue - Remove an item from the queue                       /*{{{*/
 377 // ---------------------------------------------------------------------
 378 /* We return true if we hit something*/
 379 bool pkgAcquire::Queue::Dequeue(Item *Owner)
 380 {
 381    bool Res = false;
 382
 383    QItem **I = &Items;
 384    for (; *I != 0;)
 385    {
 386       if ((*I)->Owner == Owner)
 387       {
 388          QItem *Jnk= *I;
 389          *I = (*I)->Next;
 390          Owner->QueueCounter--;
 391          delete Jnk;
 392          Res = true;
 393       }
 394       else
 395          I = &(*I)->Next;
 396    }
 397
 398    return Res;
 399 }
 400                                                                         /*}}}*/
 401 // Queue::Startup - Start the worker processes                          /*{{{*/
 402 // ---------------------------------------------------------------------
 403 /* */
 404 bool pkgAcquire::Queue::Startup()
 405 {
 406    Shutdown();
 407
 408    URI U(Name);
 409    pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access);
 410    if (Cnf == 0)
 411       return false;
 412
 413    Workers = new Worker(this,Cnf);
 414    Owner->Add(Workers);
 415    if (Workers->Start() == false)
 416       return false;
 417
 418    return Cycle();
 419 }
 420                                                                         /*}}}*/
 421 // Queue::Shutdown - Shutdown the worker processes                      /*{{{*/
 422 // ---------------------------------------------------------------------
 423 /* */
 424 bool pkgAcquire::Queue::Shutdown()
 425 {
 426    // Delete all of the workers
 427    while (Workers != 0)
 428    {
 429       pkgAcquire::Worker *Jnk = Workers;
 430       Workers = Workers->NextQueue;
 431       Owner->Remove(Jnk);
 432       delete Jnk;
 433    }
 434
 435    return true;
 436 }
 437                                                                         /*}}}*/
 438 // Queue::Finditem - Find a URI in the item list                        /*{{{*/
 439 // ---------------------------------------------------------------------
 440 /* */
 441 pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
 442 {
 443    for (QItem *I = Items; I != 0; I = I->Next)
 444       if (I->URI == URI && I->Worker == Owner)
 445          return I;
 446    return 0;
 447 }
 448                                                                         /*}}}*/
 449 // Queue::ItemDone - Item has been completed                            /*{{{*/
 450 // ---------------------------------------------------------------------
 451 /* The worker signals this which causes the item to be removed from the
 452    queue. If this is the last queue instance then it is removed from the
 453    main queue too.*/
 454 bool pkgAcquire::Queue::ItemDone(QItem *Itm)
 455 {
 456    if (Itm->Owner->QueueCounter <= 1)
 457       Owner->Dequeue(Itm->Owner);
 458    else
 459    {
 460       Dequeue(Itm->Owner);
 461       Owner->Bump();
 462    }
 463
 464    return Cycle();
 465 }
 466                                                                         /*}}}*/
 467 // Queue::Cycle - Queue new items into the method                       /*{{{*/
 468 // ---------------------------------------------------------------------
 469 /* This locates a new idle item and sends it to the worker */
 470 bool pkgAcquire::Queue::Cycle()
 471 {
 472    if (Items == 0 || Workers == 0)
 473       return true;
 474
 475    cout << "Cylce for " << Name << endl;
 476
 477    // Look for a queable item
 478    QItem *I = Items;
 479    for (; I != 0; I = I->Next)
 480       if (I->Owner->Status == pkgAcquire::Item::StatIdle)
 481          break;
 482
 483    // Nothing to do, queue is idle.
 484    if (I == 0)
 485       return true;
 486
 487    I->Worker = Workers;
 488    I->Owner->Status = pkgAcquire::Item::StatFetching;
 489    cout << "Item has been queued!" << endl;
 490    return Workers->QueueItem(I);
 491 }
 492                                                                         /*}}}*/
 493 // Queue::Bump - Fetch any pending objects if we are idle               /*{{{*/
 494 // ---------------------------------------------------------------------
 495 /* */
 496 void pkgAcquire::Queue::Bump()
 497 {
 498 }
 499                                                                         /*}}}*/