From 924aaedfcef5a0a209c8b7579ca0a77ace3102d4 Mon Sep 17 00:00:00 2001 From: UbitUmarov Date: Thu, 1 Sep 2016 12:45:09 +0100 Subject: [PATCH] assets uploads to grid. Change the retry code. Stop using asset.UploadAttempts field that should be removed --- OpenSim/Data/Tests/AssetTests.cs | 18 -- .../Asset/AssetServicesConnector.cs | 204 +++++++++++------- 2 files changed, 127 insertions(+), 95 deletions(-) diff --git a/OpenSim/Data/Tests/AssetTests.cs b/OpenSim/Data/Tests/AssetTests.cs index 5982a091df..5d7b16962f 100644 --- a/OpenSim/Data/Tests/AssetTests.cs +++ b/OpenSim/Data/Tests/AssetTests.cs @@ -125,20 +125,14 @@ namespace OpenSim.Data.Tests m_db.StoreAsset(a1); m_db.StoreAsset(a2); m_db.StoreAsset(a3); - a1.UploadAttempts = 0; - a2.UploadAttempts = 0; - a3.UploadAttempts = 0; AssetBase a1a = m_db.GetAsset(uuid1); - a1a.UploadAttempts = 0; Assert.That(a1a, Constraints.PropertyCompareConstraint(a1)); AssetBase a2a = m_db.GetAsset(uuid2); - a2a.UploadAttempts = 0; Assert.That(a2a, Constraints.PropertyCompareConstraint(a2)); AssetBase a3a = m_db.GetAsset(uuid3); - a3a.UploadAttempts = 0; Assert.That(a3a, Constraints.PropertyCompareConstraint(a3)); scrambler.Scramble(a1a); @@ -148,20 +142,14 @@ namespace OpenSim.Data.Tests m_db.StoreAsset(a1a); m_db.StoreAsset(a2a); m_db.StoreAsset(a3a); - a1a.UploadAttempts = 0; - a2a.UploadAttempts = 0; - a3a.UploadAttempts = 0; AssetBase a1b = m_db.GetAsset(uuid1); - a1b.UploadAttempts = 0; Assert.That(a1b, Constraints.PropertyCompareConstraint(a1a)); AssetBase a2b = m_db.GetAsset(uuid2); - a2b.UploadAttempts = 0; Assert.That(a2b, Constraints.PropertyCompareConstraint(a2a)); AssetBase a3b = m_db.GetAsset(uuid3); - a3b.UploadAttempts = 0; Assert.That(a3b, Constraints.PropertyCompareConstraint(a3a)); bool[] exist = m_db.AssetsExist(new[] { uuid1, uuid2, uuid3 }); @@ -202,22 +190,16 @@ namespace OpenSim.Data.Tests a3.Data = data1; m_db.StoreAsset(a1); - a1.UploadAttempts = 0; m_db.StoreAsset(a2); - a2.UploadAttempts = 0; m_db.StoreAsset(a3); - a3.UploadAttempts = 0; AssetBase a1a = m_db.GetAsset(uuid1); - a1a.UploadAttempts = 0; Assert.That(a1a, Constraints.PropertyCompareConstraint(a1)); AssetBase a2a = m_db.GetAsset(uuid2); - a2a.UploadAttempts = 0; Assert.That(a2a, Constraints.PropertyCompareConstraint(a2)); AssetBase a3a = m_db.GetAsset(uuid3); - a3a.UploadAttempts = 0; Assert.That(a3a, Constraints.PropertyCompareConstraint(a3)); } } diff --git a/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs b/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs index 7ac7917a29..5e867715bc 100644 --- a/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs +++ b/OpenSim/Services/Connectors/Asset/AssetServicesConnector.cs @@ -46,10 +46,12 @@ namespace OpenSim.Services.Connectors LogManager.GetLogger( MethodBase.GetCurrentMethod().DeclaringType); +// const int MAXSENDRETRIESLEN = 30; + const int MAXSENDRETRIESLEN = 2; private string m_ServerURI = String.Empty; private IImprovedAssetCache m_Cache = null; private int m_retryCounter; - private Dictionary> m_retryQueue = new Dictionary>(); + private List[] m_sendRetries = new List[MAXSENDRETRIESLEN]; private System.Timers.Timer m_retryTimer; private int m_maxAssetRequestConcurrency = 30; @@ -110,9 +112,9 @@ namespace OpenSim.Services.Connectors throw new Exception("Asset connector init error"); } - m_retryTimer = new System.Timers.Timer(); m_retryTimer.Elapsed += new ElapsedEventHandler(retryCheck); + m_retryTimer.AutoReset = false; m_retryTimer.Interval = 60000; Uri serverUri = new Uri(m_ServerURI); @@ -167,47 +169,57 @@ namespace OpenSim.Services.Connectors protected void retryCheck(object source, ElapsedEventArgs e) { m_retryCounter++; - if (m_retryCounter > 60) - m_retryCounter -= 60; + if(m_retryCounter >= 61 ) // avoid overflow 60 is max in use below + m_retryCounter = 1; - List keys = new List(); - foreach (int a in m_retryQueue.Keys) - { - keys.Add(a); - } - foreach (int a in keys) + int inUse = 0; + int nextlevel; + int timefactor; + List retrylist; + // we need to go down + for(int i = MAXSENDRETRIESLEN - 1; i >= 0; i--) { + lock(m_sendRetries) + retrylist = m_sendRetries[i]; + + if(retrylist == null) + continue; + + inUse++; + nextlevel = i + 1; + //We exponentially fall back on frequency until we reach one attempt per hour //The net result is that we end up in the queue for roughly 24 hours.. //24 hours worth of assets could be a lot, so the hope is that the region admin //will have gotten the asset connector back online quickly! - - int timefactor = a ^ 2; - if (timefactor > 60) + if(i == 0) + timefactor = 1; + else { - timefactor = 60; + timefactor = 1 << nextlevel; + if (timefactor > 60) + timefactor = 60; } - //First, find out if we care about this timefactor - if (timefactor % a == 0) - { - //Yes, we do! - List retrylist = m_retryQueue[a]; - m_retryQueue.Remove(a); + if(m_retryCounter < timefactor) + continue; // to update inUse; - foreach(AssetBase ass in retrylist) - { - Store(ass); //Store my ass. This function will put it back in the dictionary if it fails - } - } + if (m_retryCounter % timefactor != 0) + continue; + + // a list to retry + lock(m_sendRetries) + m_sendRetries[i] = null; + + // we are the only ones with a copy of this retrylist now + foreach(AssetBase ass in retrylist) + retryStore(ass, nextlevel); } - if (m_retryQueue.Count == 0) - { - //It might only be one tick per minute, but I have - //repented and abandoned my wasteful ways - m_retryCounter = 0; - m_retryTimer.Stop(); + lock(m_sendRetries) + { + if(inUse > 0 && !m_retryTimer.Enabled) + m_retryTimer.Start(); } } @@ -237,8 +249,9 @@ namespace OpenSim.Services.Connectors asset = SynchronousRestObjectRequester.MakeRequest("GET", uri, 0, m_Auth); - if (m_Cache != null) - m_Cache.Cache(asset); + + if (asset != null && m_Cache != null) + m_Cache.Cache(asset); } return asset; } @@ -340,25 +353,18 @@ namespace OpenSim.Services.Connectors m_AssetHandlers.Remove(id); } - Util.FireAndForget(x => + if(handlers != null) + { + Util.FireAndForget(x => { - foreach (AssetRetrievedEx h in handlers) { - // Util.FireAndForget(x => - // { try { h.Invoke(a); } catch { } - // }); } - - if (handlers != null) - handlers.Clear(); - + handlers.Clear(); }); - -// if (handlers != null) -// handlers.Clear(); + } success = true; } } @@ -393,29 +399,24 @@ namespace OpenSim.Services.Connectors { AssetRetrievedEx handlerEx = new AssetRetrievedEx(delegate(AssetBase _asset) { handler(id, sender, _asset); }); -// AssetRetrievedEx handlers; List handlers; if (m_AssetHandlers.TryGetValue(id, out handlers)) { // Someone else is already loading this asset. It will notify our handler when done. -// handlers += handlerEx; handlers.Add(handlerEx); return true; } - // Load the asset ourselves -// handlers += handlerEx; handlers = new List(); handlers.Add(handlerEx); m_AssetHandlers.Add(id, handlers); + + QueuedAssetRequest request = new QueuedAssetRequest(); + request.id = id; + request.uri = uri; + m_requestQueue.Enqueue(request); } - - QueuedAssetRequest request = new QueuedAssetRequest(); - request.id = id; - request.uri = uri; - - m_requestQueue.Enqueue(request); } else { @@ -495,43 +496,35 @@ namespace OpenSim.Services.Connectors newID = SynchronousRestObjectRequester. MakeRequest("POST", uri, asset, 100000, m_Auth); } - catch {} + catch + { + newID = null; + } if (newID == null || newID == String.Empty || newID == stringUUIDZero) { - //The asset upload failed, put it in a queue for later - asset.UploadAttempts++; - if (asset.UploadAttempts > 30) + //The asset upload failed, try later + lock(m_sendRetries) { - //By this stage we've been in the queue for a good few hours; - //We're going to drop the asset. - m_log.ErrorFormat("[Assets] Dropping asset {0} - Upload has been in the queue for too long.", asset.ID.ToString()); - } - else - { - if (!m_retryQueue.ContainsKey(asset.UploadAttempts)) - { - m_retryQueue.Add(asset.UploadAttempts, new List()); - } - List m_queue = m_retryQueue[asset.UploadAttempts]; + if (m_sendRetries[0] == null) + m_sendRetries[0] = new List(); + List m_queue = m_sendRetries[0]; m_queue.Add(asset); - m_log.WarnFormat("[Assets] Upload failed: {0} - Requeuing asset for another run.", asset.ID.ToString()); - m_retryTimer.Start(); + m_log.WarnFormat("[Assets] Upload failed: {0} type {1} will retry later", + asset.ID.ToString(), asset.Type.ToString()); + if(!m_retryTimer.Enabled) + m_retryTimer.Start(); } } else { - if (asset.UploadAttempts > 0) - { - m_log.InfoFormat("[Assets] Upload of {0} succeeded after {1} failed attempts", asset.ID.ToString(), asset.UploadAttempts.ToString()); - } if (newID != asset.ID) { // Placing this here, so that this work with old asset servers that don't send any reply back // SynchronousRestObjectRequester returns somethins that is not an empty string asset.ID = newID; -// what about FullID ???? + if (m_Cache != null) m_Cache.Cache(asset); } @@ -539,6 +532,62 @@ namespace OpenSim.Services.Connectors return asset.ID; } + public void retryStore(AssetBase asset, int nextRetryLevel) + { +/* this may be bad, so excluding + if (m_Cache != null && !m_Cache.Check(asset.ID)) + { + m_log.WarnFormat("[Assets] Upload giveup asset bc no longer in local cache: {0}", + asset.ID.ToString(); + return; // if no longer in cache, it was deleted or expired + } +*/ + string uri = MapServer(asset.FullID.ToString()) + "/assets/"; + + string newID = null; + try + { + newID = SynchronousRestObjectRequester. + MakeRequest("POST", uri, asset, 100000, m_Auth); + } + catch + { + newID = null; + } + + if (newID == null || newID == String.Empty || newID == stringUUIDZero) + { + if(nextRetryLevel >= MAXSENDRETRIESLEN) + m_log.WarnFormat("[Assets] Upload giveup after several retries id: {0} type {1}", + asset.ID.ToString(), asset.Type.ToString()); + else + { + lock(m_sendRetries) + { + if (m_sendRetries[nextRetryLevel] == null) + { + m_sendRetries[nextRetryLevel] = new List(); + } + List m_queue = m_sendRetries[nextRetryLevel]; + m_queue.Add(asset); + m_log.WarnFormat("[Assets] Upload failed: {0} type {1} will retry later", + asset.ID.ToString(), asset.Type.ToString()); + } + } + } + else + { + m_log.InfoFormat("[Assets] Upload of {0} succeeded after {1} failed attempts", asset.ID.ToString(), nextRetryLevel.ToString()); + if (newID != asset.ID) + { + asset.ID = newID; + + if (m_Cache != null) + m_Cache.Cache(asset); + } + } + } + public bool UpdateContent(string id, byte[] data) { AssetBase asset = null; @@ -569,6 +618,7 @@ namespace OpenSim.Services.Connectors return false; } + public bool Delete(string id) { string uri = MapServer(id) + "/assets/" + id;