acquire.cc 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420
  1. // -*- mode: cpp; mode: fold -*-
  2. // Description /*{{{*/
  3. // $Id: acquire.cc,v 1.4 1998/10/24 04:58:01 jgg Exp $
  4. /* ######################################################################
  5. Acquire - File Acquiration
  6. The core element for the schedual system is the concept of a named
  7. queue. Each queue is unique and each queue has a name derived from the
  8. URI. The degree of paralization can be controled by how the queue
  9. name is derived from the URI.
  10. ##################################################################### */
  11. /*}}}*/
  12. // Include Files /*{{{*/
  13. #ifdef __GNUG__
  14. #pragma implementation "apt-pkg/acquire.h"
  15. #endif
  16. #include <apt-pkg/acquire.h>
  17. #include <apt-pkg/acquire-item.h>
  18. #include <apt-pkg/acquire-worker.h>
  19. #include <apt-pkg/configuration.h>
  20. #include <apt-pkg/error.h>
  21. #include <strutl.h>
  22. /*}}}*/
  23. // Acquire::pkgAcquire - Constructor /*{{{*/
  24. // ---------------------------------------------------------------------
  25. /* */
  26. pkgAcquire::pkgAcquire()
  27. {
  28. Queues = 0;
  29. Configs = 0;
  30. Workers = 0;
  31. ToFetch = 0;
  32. string Mode = _config->Find("Acquire::Queue-Mode","host");
  33. if (strcasecmp(Mode.c_str(),"host") == 0)
  34. QueueMode = QueueHost;
  35. if (strcasecmp(Mode.c_str(),"access") == 0)
  36. QueueMode = QueueAccess;
  37. Debug = _config->FindB("Debug::pkgAcquire",false);
  38. }
  39. /*}}}*/
  40. // Acquire::~pkgAcquire - Destructor /*{{{*/
  41. // ---------------------------------------------------------------------
  42. /* Free our memory */
  43. pkgAcquire::~pkgAcquire()
  44. {
  45. while (Items.size() != 0)
  46. delete Items[0];
  47. while (Configs != 0)
  48. {
  49. MethodConfig *Jnk = Configs;
  50. Configs = Configs->Next;
  51. delete Jnk;
  52. }
  53. while (Queues != 0)
  54. {
  55. Queue *Jnk = Queues;
  56. Queues = Queues->Next;
  57. delete Jnk;
  58. }
  59. }
  60. /*}}}*/
  61. // Acquire::Add - Add a new item /*{{{*/
  62. // ---------------------------------------------------------------------
  63. /* */
  64. void pkgAcquire::Add(Item *Itm)
  65. {
  66. Items.push_back(Itm);
  67. }
  68. /*}}}*/
  69. // Acquire::Remove - Remove a item /*{{{*/
  70. // ---------------------------------------------------------------------
  71. /* */
  72. void pkgAcquire::Remove(Item *Itm)
  73. {
  74. for (vector<Item *>::iterator I = Items.begin(); I < Items.end(); I++)
  75. {
  76. if (*I == Itm)
  77. Items.erase(I);
  78. }
  79. }
  80. /*}}}*/
  81. // Acquire::Add - Add a worker /*{{{*/
  82. // ---------------------------------------------------------------------
  83. /* */
  84. void pkgAcquire::Add(Worker *Work)
  85. {
  86. Work->NextAcquire = Workers;
  87. Workers = Work;
  88. }
  89. /*}}}*/
  90. // Acquire::Remove - Remove a worker /*{{{*/
  91. // ---------------------------------------------------------------------
  92. /* */
  93. void pkgAcquire::Remove(Worker *Work)
  94. {
  95. Worker **I = &Workers;
  96. for (; *I != 0;)
  97. {
  98. if (*I == Work)
  99. *I = (*I)->NextAcquire;
  100. else
  101. I = &(*I)->NextAcquire;
  102. }
  103. }
  104. /*}}}*/
  105. // Acquire::Enqueue - Queue an URI for fetching /*{{{*/
  106. // ---------------------------------------------------------------------
  107. /* */
  108. void pkgAcquire::Enqueue(Item *Itm,string URI,string Description)
  109. {
  110. // Determine which queue to put the item in
  111. string Name = QueueName(URI);
  112. if (Name.empty() == true)
  113. return;
  114. // Find the queue structure
  115. Queue *I = Queues;
  116. for (; I != 0 && I->Name != Name; I = I->Next);
  117. if (I == 0)
  118. {
  119. I = new Queue(Name,this);
  120. I->Next = Queues;
  121. Queues = I;
  122. }
  123. // Queue it into the named queue
  124. I->Enqueue(Itm,URI,Description);
  125. ToFetch++;
  126. // Some trace stuff
  127. if (Debug == true)
  128. {
  129. clog << "Fetching " << URI << endl;
  130. clog << " to " << Itm->DestFile << endl;
  131. clog << " Queue is: " << QueueName(URI) << endl;
  132. }
  133. }
  134. /*}}}*/
  135. // Acquire::Dequeue - Remove an item from all queues /*{{{*/
  136. // ---------------------------------------------------------------------
  137. /* */
  138. void pkgAcquire::Dequeue(Item *Itm)
  139. {
  140. Queue *I = Queues;
  141. for (; I != 0; I = I->Next)
  142. I->Dequeue(Itm);
  143. ToFetch--;
  144. }
  145. /*}}}*/
  146. // Acquire::QueueName - Return the name of the queue for this URI /*{{{*/
  147. // ---------------------------------------------------------------------
  148. /* The string returned depends on the configuration settings and the
  149. method parameters. Given something like http://foo.org/bar it can
  150. return http://foo.org or http */
  151. string pkgAcquire::QueueName(string URI)
  152. {
  153. const MethodConfig *Config = GetConfig(URIAccess(URI));
  154. if (Config == 0)
  155. return string();
  156. /* Single-Instance methods get exactly one queue per URI. This is
  157. also used for the Access queue method */
  158. if (Config->SingleInstance == true || QueueMode == QueueAccess)
  159. return URIAccess(URI);
  160. // Host based queue
  161. string::iterator I = URI.begin();
  162. for (; I < URI.end() && *I != ':'; I++);
  163. for (; I < URI.end() && (*I == '/' || *I == ':'); I++);
  164. for (; I < URI.end() && *I != '/'; I++);
  165. return string(URI,0,I - URI.begin());
  166. }
  167. /*}}}*/
  168. // Acquire::GetConfig - Fetch the configuration information /*{{{*/
  169. // ---------------------------------------------------------------------
  170. /* This locates the configuration structure for an access method. If
  171. a config structure cannot be found a Worker will be created to
  172. retrieve it */
  173. pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
  174. {
  175. // Search for an existing config
  176. MethodConfig *Conf;
  177. for (Conf = Configs; Conf != 0; Conf = Conf->Next)
  178. if (Conf->Access == Access)
  179. return Conf;
  180. // Create the new config class
  181. Conf = new MethodConfig;
  182. Conf->Access = Access;
  183. Conf->Next = Configs;
  184. Configs = Conf;
  185. // Create the worker to fetch the configuration
  186. Worker Work(Conf);
  187. if (Work.Start() == false)
  188. return 0;
  189. return Conf;
  190. }
  191. /*}}}*/
  192. // Acquire::SetFds - Deal with readable FDs /*{{{*/
  193. // ---------------------------------------------------------------------
  194. /* Collect FDs that have activity monitors into the fd sets */
  195. void pkgAcquire::SetFds(int &Fd,fd_set *RSet,fd_set *WSet)
  196. {
  197. for (Worker *I = Workers; I != 0; I = I->NextAcquire)
  198. {
  199. if (I->InReady == true && I->InFd >= 0)
  200. {
  201. if (Fd < I->InFd)
  202. Fd = I->InFd;
  203. FD_SET(I->InFd,RSet);
  204. }
  205. if (I->OutReady == true && I->OutFd >= 0)
  206. {
  207. if (Fd < I->OutFd)
  208. Fd = I->OutFd;
  209. FD_SET(I->OutFd,WSet);
  210. }
  211. }
  212. }
  213. /*}}}*/
  214. // Acquire::RunFds - Deal with active FDs /*{{{*/
  215. // ---------------------------------------------------------------------
  216. /* Dispatch active FDs over to the proper workers */
  217. void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet)
  218. {
  219. for (Worker *I = Workers; I != 0; I = I->NextAcquire)
  220. {
  221. if (I->InFd >= 0 && FD_ISSET(I->InFd,RSet) != 0)
  222. I->InFdReady();
  223. if (I->OutFd >= 0 && FD_ISSET(I->OutFd,WSet) != 0)
  224. I->OutFdReady();
  225. }
  226. }
  227. /*}}}*/
  228. // Acquire::Run - Run the fetch sequence /*{{{*/
  229. // ---------------------------------------------------------------------
  230. /* This runs the queues. It manages a select loop for all of the
  231. Worker tasks. The workers interact with the queues and items to
  232. manage the actual fetch. */
  233. bool pkgAcquire::Run()
  234. {
  235. for (Queue *I = Queues; I != 0; I = I->Next)
  236. I->Startup();
  237. // Run till all things have been acquired
  238. while (ToFetch > 0)
  239. {
  240. fd_set RFds;
  241. fd_set WFds;
  242. int Highest = 0;
  243. FD_ZERO(&RFds);
  244. FD_ZERO(&WFds);
  245. SetFds(Highest,&RFds,&WFds);
  246. if (select(Highest+1,&RFds,&WFds,0,0) <= 0)
  247. return _error->Errno("select","Select has failed");
  248. RunFds(&RFds,&WFds);
  249. }
  250. for (Queue *I = Queues; I != 0; I = I->Next)
  251. I->Shutdown();
  252. return true;
  253. }
  254. /*}}}*/
  255. // Acquire::MethodConfig::MethodConfig - Constructor /*{{{*/
  256. // ---------------------------------------------------------------------
  257. /* */
  258. pkgAcquire::MethodConfig::MethodConfig()
  259. {
  260. SingleInstance = false;
  261. PreScan = false;
  262. Pipeline = false;
  263. SendConfig = false;
  264. Next = 0;
  265. }
  266. /*}}}*/
  267. // Queue::Queue - Constructor /*{{{*/
  268. // ---------------------------------------------------------------------
  269. /* */
  270. pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name),
  271. Owner(Owner)
  272. {
  273. Items = 0;
  274. Next = 0;
  275. Workers = 0;
  276. }
  277. /*}}}*/
  278. // Queue::~Queue - Destructor /*{{{*/
  279. // ---------------------------------------------------------------------
  280. /* */
  281. pkgAcquire::Queue::~Queue()
  282. {
  283. Shutdown();
  284. while (Items != 0)
  285. {
  286. QItem *Jnk = Items;
  287. Items = Items->Next;
  288. delete Jnk;
  289. }
  290. }
  291. /*}}}*/
  292. // Queue::Enqueue - Queue an item to the queue /*{{{*/
  293. // ---------------------------------------------------------------------
  294. /* */
  295. void pkgAcquire::Queue::Enqueue(Item *Owner,string URI,string Description)
  296. {
  297. // Create a new item
  298. QItem *I = new QItem;
  299. I->Next = Items;
  300. Items = I;
  301. // Fill it in
  302. Items->Owner = Owner;
  303. Items->URI = URI;
  304. Items->Description = Description;
  305. Owner->QueueCounter++;
  306. }
  307. /*}}}*/
  308. // Queue::Dequeue - Remove an item from the queue /*{{{*/
  309. // ---------------------------------------------------------------------
  310. /* */
  311. void pkgAcquire::Queue::Dequeue(Item *Owner)
  312. {
  313. QItem **I = &Items;
  314. for (; *I != 0;)
  315. {
  316. if ((*I)->Owner == Owner)
  317. {
  318. QItem *Jnk= *I;
  319. *I = (*I)->Next;
  320. Owner->QueueCounter--;
  321. delete Jnk;
  322. }
  323. else
  324. I = &(*I)->Next;
  325. }
  326. }
  327. /*}}}*/
  328. // Queue::Startup - Start the worker processes /*{{{*/
  329. // ---------------------------------------------------------------------
  330. /* */
  331. bool pkgAcquire::Queue::Startup()
  332. {
  333. Shutdown();
  334. pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(URIAccess(Name));
  335. if (Cnf == 0)
  336. return false;
  337. Workers = new Worker(this,Cnf);
  338. Owner->Add(Workers);
  339. if (Workers->Start() == false)
  340. return false;
  341. Items->Worker = Workers;
  342. Workers->QueueItem(Items);
  343. return true;
  344. }
  345. /*}}}*/
  346. // Queue::Shutdown - Shutdown the worker processes /*{{{*/
  347. // ---------------------------------------------------------------------
  348. /* */
  349. bool pkgAcquire::Queue::Shutdown()
  350. {
  351. // Delete all of the workers
  352. while (Workers != 0)
  353. {
  354. pkgAcquire::Worker *Jnk = Workers;
  355. Workers = Workers->NextQueue;
  356. Owner->Remove(Jnk);
  357. delete Jnk;
  358. }
  359. return true;
  360. }
  361. /*}}}*/
  362. // Queue::Finditem - Find a URI in the item list /*{{{*/
  363. // ---------------------------------------------------------------------
  364. /* */
  365. pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
  366. {
  367. for (QItem *I = Items; I != 0; I = I->Next)
  368. if (I->URI == URI && I->Worker == Owner)
  369. return I;
  370. return 0;
  371. }
  372. /*}}}*/
  373. // Queue::ItemDone - Item has been completed /*{{{*/
  374. // ---------------------------------------------------------------------
  375. /* The worker signals this which causes the item to be removed from the
  376. queue. */
  377. bool pkgAcquire::Queue::ItemDone(QItem *Itm)
  378. {
  379. Dequeue(Itm->Owner);
  380. if (Items == 0)
  381. return true;
  382. Items->Worker = Workers;
  383. Items->Owner->Status = pkgAcquire::Item::StatFetching;
  384. return Workers->QueueItem(Items);
  385. }
  386. /*}}}*/