/* * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ using System; using System.Collections; using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; using System.Threading; using QuantConnect.Configuration; using QuantConnect.Interfaces; using QuantConnect.Logging; using QuantConnect.Packets; using QuantConnect.Util; namespace QuantConnect.Lean.Engine.Storage { /// /// A local disk implementation of . /// public class LocalObjectStore : IObjectStore { /// /// No read permissions error message /// protected const string NoReadPermissionsError = "The current user does not have permission to read from the organization Object Store." + " Please contact your organization administrator to request permission."; /// /// No write permissions error message /// protected const string NoWritePermissionsError = "The current user does not have permission to write to the organization Object Store." + " Please contact your organization administrator to request permission."; /// /// Event raised each time there's an error /// public event EventHandler ErrorRaised; /// /// Flag indicating the state of this object storage has changed since the last invocation /// private volatile bool _dirty; private Timer _persistenceTimer; private TimeSpan _persistenceInterval; private readonly string _storageRoot = Path.GetFullPath(Config.Get("object-store-root", "./storage")); private readonly ConcurrentDictionary _storage = new ConcurrentDictionary(); private readonly object _persistLock = new object(); /// /// Provides access to the controls governing behavior of this instance, such as the persistence interval /// protected Controls Controls { get; private set; } /// /// The root storage folder for the algorithm /// protected string AlgorithmStorageRoot { get; private set; } /// /// Initializes the object store /// /// The algorithm name /// The user id /// The project id /// The user token /// The job controls instance public virtual void Initialize(string algorithmName, int userId, int projectId, string userToken, Controls controls) { // absolute path including algorithm name AlgorithmStorageRoot = Path.Combine(_storageRoot, algorithmName); // create the root path if it does not exist Directory.CreateDirectory(AlgorithmStorageRoot); Log.Trace($"LocalObjectStore.Initialize(): Storage Root: {new FileInfo(AlgorithmStorageRoot).FullName}"); Controls = controls; // Load in any already existing objects in the storage directory LoadExistingObjects(); // if <= 0 we disable periodic persistence and make it synchronous if (Controls.PersistenceIntervalSeconds > 0) { _persistenceInterval = TimeSpan.FromSeconds(Controls.PersistenceIntervalSeconds); _persistenceTimer = new Timer(_ => Persist(), null, _persistenceInterval, _persistenceInterval); } } /// /// Loads objects from the AlgorithmStorageRoot into the ObjectStore /// protected virtual void LoadExistingObjects() { if (Controls.StoragePermissions.HasFlag(FileAccess.Read)) { foreach (var file in Directory.EnumerateFiles(AlgorithmStorageRoot)) { // Read the contents of the file and decode the filename to get our key var contents = File.ReadAllBytes(file); var key = Base64ToKey(Path.GetFileName(file)); _storage[key] = contents; } } } /// /// Determines whether the store contains data for the specified key /// /// The object key /// True if the key was found public bool ContainsKey(string key) { if (key == null) { throw new ArgumentNullException(nameof(key)); } if (!Controls.StoragePermissions.HasFlag(FileAccess.Read)) { throw new InvalidOperationException($"LocalObjectStore.ContainsKey(): {NoReadPermissionsError}"); } return _storage.ContainsKey(key); } /// /// Returns the object data for the specified key /// /// The object key /// A byte array containing the data public byte[] ReadBytes(string key) { // Ensure we have the key, also takes care of null or improper access if (!ContainsKey(key)) { throw new KeyNotFoundException($"Object with key '{key}' was not found in the current project. " + "Please use ObjectStore.ContainsKey(key) to check if an object exists before attempting to read." ); } byte[] data; _storage.TryGetValue(key, out data); return data; } /// /// Saves the object data for the specified key /// /// The object key /// The object data /// True if the save operation was successful public bool SaveBytes(string key, byte[] contents) { if (key == null) { throw new ArgumentNullException(nameof(key)); } if (key.Contains("?")) { throw new ArgumentException($"LocalObjectStore.SaveBytes(): char '?' is not supported in the key {key}"); } if (!Controls.StoragePermissions.HasFlag(FileAccess.Write)) { throw new InvalidOperationException($"LocalObjectStore.SaveBytes(): {NoWritePermissionsError}"); } if (InternalSaveBytes(key, contents)) { _dirty = true; // if <= 0 we disable periodic persistence and make it synchronous if (Controls.PersistenceIntervalSeconds <= 0) { Persist(); } return true; } return false; } /// /// Won't trigger persist nor will check storage write permissions, useful on initialization since it allows read only permissions to load the object store /// protected bool InternalSaveBytes(string key, byte[] contents) { // Before saving confirm we are abiding by the control rules // Start by counting our file and its length var fileCount = 1; var expectedStorageSizeBytes = contents.Length; foreach (var kvp in _storage) { if (key.Equals(kvp.Key)) { // Skip we have already counted this above // If this key was already in storage it will be replaced. } else { fileCount++; expectedStorageSizeBytes += kvp.Value.Length; } } // Verify we are within FileCount limit if (fileCount > Controls.StorageFileCount) { var message = $"LocalObjectStore.InternalSaveBytes(): at file capacity: {fileCount}. Unable to save: '{key}'"; Log.Error(message); OnErrorRaised(new StorageLimitExceededException(message)); return false; } // Verify we are within Storage limit var expectedStorageSizeMb = BytesToMb(expectedStorageSizeBytes); if (expectedStorageSizeMb > Controls.StorageLimitMB) { var message = $"LocalObjectStore.InternalSaveBytes(): at storage capacity: {expectedStorageSizeMb}MB/{Controls.StorageLimitMB}MB. Unable to save: '{key}'"; Log.Error(message); OnErrorRaised(new StorageLimitExceededException(message)); return false; } // Add the entry _storage.AddOrUpdate(key, k => contents, (k, v) => contents); return true; } /// /// Deletes the object data for the specified key /// /// The object key /// True if the delete operation was successful public bool Delete(string key) { if (key == null) { throw new ArgumentNullException(nameof(key)); } if (!Controls.StoragePermissions.HasFlag(FileAccess.Write)) { throw new InvalidOperationException($"LocalObjectStore.Delete(): {NoWritePermissionsError}"); } byte[] _; if (_storage.TryRemove(key, out _)) { _dirty = true; // if <= 0 we disable periodic persistence and make it synchronous if (Controls.PersistenceIntervalSeconds <= 0) { Persist(); } return true; } return false; } /// /// Returns the file path for the specified key /// /// The object key /// The path for the file public virtual string GetFilePath(string key) { // Ensure we have an object for that key if (!ContainsKey(key)) { throw new KeyNotFoundException($"Object with key '{key}' was not found in the current project. " + "Please use ObjectStore.ContainsKey(key) to check if an object exists before attempting to read." ); } // Persist to ensure pur files are up to date Persist(); // Fetch the path to file and return it return PathForKey(key); } /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// public virtual void Dispose() { try { if (_persistenceTimer != null) { _persistenceTimer.Change(Timeout.Infinite, Timeout.Infinite); Persist(); _persistenceTimer.DisposeSafely(); } // if the object store was not used, delete the empty storage directory created in Initialize. if (AlgorithmStorageRoot != null && !Directory.GetFileSystemEntries(AlgorithmStorageRoot).Any()) { Directory.Delete(AlgorithmStorageRoot); } } catch (Exception err) { Log.Error(err, "Error deleting storage directory."); } } /// Returns an enumerator that iterates through the collection. /// A that can be used to iterate through the collection. /// 1 public IEnumerator> GetEnumerator() { return _storage.GetEnumerator(); } /// Returns an enumerator that iterates through a collection. /// An object that can be used to iterate through the collection. /// 2 IEnumerator IEnumerable.GetEnumerator() { return GetEnumerator(); } /// /// Get's a file path for a given key. /// Internal use only because it does not guarantee the existence of the file. /// protected string PathForKey(string key) { // We use an encoded filename because certain keys will cause problems with persisting // data to a file; we use Base64 because it allow us to use all chars except '?' in our // key, this is because '?' in the right place will output '/' which will break during // persist return Path.Combine(AlgorithmStorageRoot, $"{KeyToBase64(key)}"); } /// /// Invoked periodically to persist the object store's contents /// private void Persist() { // Acquire the persist lock lock (_persistLock) { // If there are no changes we are fine if (!_dirty) { return; } try { // Pause timer while persisting _persistenceTimer?.Change(Timeout.Infinite, Timeout.Infinite); if (PersistData(this)) { _dirty = false; } } catch (Exception err) { Log.Error("LocalObjectStore.Persist()", err); OnErrorRaised(err); } finally { // restart timer following end of persistence _persistenceTimer?.Change(_persistenceInterval, _persistenceInterval); } } } /// /// Overridable persistence function /// /// The data to be persisted /// True if persistence was successful, otherwise false protected virtual bool PersistData(IEnumerable> data) { try { // Delete any files that are no longer saved in the store foreach (var filepath in Directory.EnumerateFiles(AlgorithmStorageRoot)) { var filename = Path.GetFileName(filepath); if (!_storage.ContainsKey(Base64ToKey(filename))) { File.Delete(filepath); } } // Write all our store data to disk foreach (var kvp in data) { // Get a path for this key and write to it var path = PathForKey(kvp.Key); File.WriteAllBytes(path, kvp.Value); } return true; } catch (Exception err) { Log.Error("LocalObjectStore.PersistData()", err); OnErrorRaised(err); return false; } } /// /// Event invocator for the event /// protected virtual void OnErrorRaised(Exception error) { ErrorRaised?.Invoke(this, new ObjectStoreErrorRaisedEventArgs(error)); } /// /// Converts a number of bytes to megabytes as it's more human legible /// private static double BytesToMb(long bytes) { return bytes / 1024.0 / 1024.0; } /// /// Convert a given key to a Base64 string /// /// Key to be encoded /// Base64 hash string private static string KeyToBase64(string key) { var textAsBytes = Encoding.UTF8.GetBytes(key); return Convert.ToBase64String(textAsBytes); } /// /// Convert a given Base64 string back to a key /// /// Hash to be decoded /// Key string private static string Base64ToKey(string hash) { var textAsBytes = Convert.FromBase64String(hash); return Encoding.UTF8.GetString(textAsBytes); } } }