acquire.cc 11 KB

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