/*
* 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);
}
}
}