diff options
Sequence/throttle asset retrievals.
Diffstat (limited to '')
-rw-r--r-- | OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs | 104 |
1 files changed, 67 insertions, 37 deletions
diff --git a/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs b/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs index daf38bc..9d6d9ad 100644 --- a/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs +++ b/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs | |||
@@ -27,6 +27,7 @@ | |||
27 | 27 | ||
28 | using log4net; | 28 | using log4net; |
29 | using System; | 29 | using System; |
30 | using System.Threading; | ||
30 | using System.Collections.Generic; | 31 | using System.Collections.Generic; |
31 | using System.IO; | 32 | using System.IO; |
32 | using System.Reflection; | 33 | using System.Reflection; |
@@ -50,7 +51,7 @@ namespace OpenSim.Services.Connectors | |||
50 | private IImprovedAssetCache m_Cache = null; | 51 | private IImprovedAssetCache m_Cache = null; |
51 | private int m_retryCounter; | 52 | private int m_retryCounter; |
52 | private Dictionary<int, List<AssetBase>> m_retryQueue = new Dictionary<int, List<AssetBase>>(); | 53 | private Dictionary<int, List<AssetBase>> m_retryQueue = new Dictionary<int, List<AssetBase>>(); |
53 | private Timer m_retryTimer; | 54 | private System.Timers.Timer m_retryTimer; |
54 | private delegate void AssetRetrievedEx(AssetBase asset); | 55 | private delegate void AssetRetrievedEx(AssetBase asset); |
55 | 56 | ||
56 | // Keeps track of concurrent requests for the same asset, so that it's only loaded once. | 57 | // Keeps track of concurrent requests for the same asset, so that it's only loaded once. |
@@ -61,6 +62,8 @@ namespace OpenSim.Services.Connectors | |||
61 | 62 | ||
62 | private Dictionary<string, string> m_UriMap = new Dictionary<string, string>(); | 63 | private Dictionary<string, string> m_UriMap = new Dictionary<string, string>(); |
63 | 64 | ||
65 | private Thread[] m_fetchThreads; | ||
66 | |||
64 | public AssetServicesConnector() | 67 | public AssetServicesConnector() |
65 | { | 68 | { |
66 | } | 69 | } |
@@ -96,7 +99,7 @@ namespace OpenSim.Services.Connectors | |||
96 | } | 99 | } |
97 | 100 | ||
98 | 101 | ||
99 | m_retryTimer = new Timer(); | 102 | m_retryTimer = new System.Timers.Timer(); |
100 | m_retryTimer.Elapsed += new ElapsedEventHandler(retryCheck); | 103 | m_retryTimer.Elapsed += new ElapsedEventHandler(retryCheck); |
101 | m_retryTimer.Interval = 60000; | 104 | m_retryTimer.Interval = 60000; |
102 | 105 | ||
@@ -112,6 +115,14 @@ namespace OpenSim.Services.Connectors | |||
112 | m_UriMap[prefix] = groupHost; | 115 | m_UriMap[prefix] = groupHost; |
113 | //m_log.DebugFormat("[ASSET]: Using {0} for prefix {1}", groupHost, prefix); | 116 | //m_log.DebugFormat("[ASSET]: Using {0} for prefix {1}", groupHost, prefix); |
114 | } | 117 | } |
118 | |||
119 | m_fetchThreads = new Thread[2]; | ||
120 | |||
121 | for (int i = 0 ; i < 2 ; i++) | ||
122 | { | ||
123 | m_fetchThreads[i] = new Thread(AssetRequestProcessor); | ||
124 | m_fetchThreads[i].Start(); | ||
125 | } | ||
115 | } | 126 | } |
116 | 127 | ||
117 | private string MapServer(string id) | 128 | private string MapServer(string id) |
@@ -261,37 +272,25 @@ namespace OpenSim.Services.Connectors | |||
261 | return null; | 272 | return null; |
262 | } | 273 | } |
263 | 274 | ||
264 | public bool Get(string id, Object sender, AssetRetrieved handler) | 275 | private class QueuedAssetRequest |
265 | { | 276 | { |
266 | string uri = MapServer(id) + "/assets/" + id; | 277 | public string uri; |
267 | 278 | public string id; | |
268 | AssetBase asset = null; | 279 | } |
269 | if (m_Cache != null) | ||
270 | asset = m_Cache.Get(id); | ||
271 | 280 | ||
272 | if (asset == null || asset.Data == null || asset.Data.Length == 0) | 281 | private OpenMetaverse.BlockingQueue<QueuedAssetRequest> m_requestQueue = |
273 | { | 282 | new OpenMetaverse.BlockingQueue<QueuedAssetRequest>(); |
274 | lock (m_AssetHandlers) | ||
275 | { | ||
276 | AssetRetrievedEx handlerEx = new AssetRetrievedEx(delegate(AssetBase _asset) { handler(id, sender, _asset); }); | ||
277 | 283 | ||
278 | // AssetRetrievedEx handlers; | 284 | private void AssetRequestProcessor() |
279 | List<AssetRetrievedEx> handlers; | 285 | { |
280 | if (m_AssetHandlers.TryGetValue(id, out handlers)) | 286 | QueuedAssetRequest r; |
281 | { | ||
282 | // Someone else is already loading this asset. It will notify our handler when done. | ||
283 | // handlers += handlerEx; | ||
284 | handlers.Add(handlerEx); | ||
285 | return true; | ||
286 | } | ||
287 | 287 | ||
288 | // Load the asset ourselves | 288 | while (true) |
289 | // handlers += handlerEx; | 289 | { |
290 | handlers = new List<AssetRetrievedEx>(); | 290 | r = m_requestQueue.Dequeue(); |
291 | handlers.Add(handlerEx); | ||
292 | 291 | ||
293 | m_AssetHandlers.Add(id, handlers); | 292 | string uri = r.uri; |
294 | } | 293 | string id = r.id; |
295 | 294 | ||
296 | bool success = false; | 295 | bool success = false; |
297 | try | 296 | try |
@@ -301,16 +300,7 @@ namespace OpenSim.Services.Connectors | |||
301 | { | 300 | { |
302 | if (m_Cache != null) | 301 | if (m_Cache != null) |
303 | m_Cache.Cache(a); | 302 | m_Cache.Cache(a); |
304 | /* | ||
305 | AssetRetrievedEx handlers; | ||
306 | lock (m_AssetHandlers) | ||
307 | { | ||
308 | handlers = m_AssetHandlers[id]; | ||
309 | m_AssetHandlers.Remove(id); | ||
310 | } | ||
311 | 303 | ||
312 | handlers.Invoke(a); | ||
313 | */ | ||
314 | List<AssetRetrievedEx> handlers; | 304 | List<AssetRetrievedEx> handlers; |
315 | lock (m_AssetHandlers) | 305 | lock (m_AssetHandlers) |
316 | { | 306 | { |
@@ -340,6 +330,46 @@ namespace OpenSim.Services.Connectors | |||
340 | } | 330 | } |
341 | } | 331 | } |
342 | } | 332 | } |
333 | } | ||
334 | |||
335 | public bool Get(string id, Object sender, AssetRetrieved handler) | ||
336 | { | ||
337 | string uri = MapServer(id) + "/assets/" + id; | ||
338 | |||
339 | AssetBase asset = null; | ||
340 | if (m_Cache != null) | ||
341 | asset = m_Cache.Get(id); | ||
342 | |||
343 | if (asset == null || asset.Data == null || asset.Data.Length == 0) | ||
344 | { | ||
345 | lock (m_AssetHandlers) | ||
346 | { | ||
347 | AssetRetrievedEx handlerEx = new AssetRetrievedEx(delegate(AssetBase _asset) { handler(id, sender, _asset); }); | ||
348 | |||
349 | // AssetRetrievedEx handlers; | ||
350 | List<AssetRetrievedEx> handlers; | ||
351 | if (m_AssetHandlers.TryGetValue(id, out handlers)) | ||
352 | { | ||
353 | // Someone else is already loading this asset. It will notify our handler when done. | ||
354 | // handlers += handlerEx; | ||
355 | handlers.Add(handlerEx); | ||
356 | return true; | ||
357 | } | ||
358 | |||
359 | // Load the asset ourselves | ||
360 | // handlers += handlerEx; | ||
361 | handlers = new List<AssetRetrievedEx>(); | ||
362 | handlers.Add(handlerEx); | ||
363 | |||
364 | m_AssetHandlers.Add(id, handlers); | ||
365 | } | ||
366 | |||
367 | QueuedAssetRequest request = new QueuedAssetRequest(); | ||
368 | request.id = id; | ||
369 | request.uri = uri; | ||
370 | |||
371 | m_requestQueue.Enqueue(request); | ||
372 | } | ||
343 | else | 373 | else |
344 | { | 374 | { |
345 | handler(id, sender, asset); | 375 | handler(id, sender, asset); |