Initialize the sharded LMDB manager. Args: cache_dir: Cache directory path. num_shards: Number of shards to use. map_size_per_shard: Maximum size per shard in bytes. readonly: Whether to open in read-only mode. lock: Whether to use
(
self,
cache_dir: str,
num_shards: int = 32,
map_size_per_shard: int = 1024 * 1024 * 1024 * 1024, # 1TB per shard
readonly: bool = False,
lock: bool = True,
logger=None,
max_open_shards: int = 0, # 0 means no limit
)
| 105 | _instances: Dict[str, "ShardedLMDBManager"] = {} |
| 106 | |
| 107 | def __init__( |
| 108 | self, |
| 109 | cache_dir: str, |
| 110 | num_shards: int = 32, |
| 111 | map_size_per_shard: int = 1024 * 1024 * 1024 * 1024, # 1TB per shard |
| 112 | readonly: bool = False, |
| 113 | lock: bool = True, |
| 114 | logger=None, |
| 115 | max_open_shards: int = 0, # 0 means no limit |
| 116 | ): |
| 117 | """Initialize the sharded LMDB manager. |
| 118 | |
| 119 | Args: |
| 120 | cache_dir: Cache directory path. |
| 121 | num_shards: Number of shards to use. |
| 122 | map_size_per_shard: Maximum size per shard in bytes. |
| 123 | readonly: Whether to open in read-only mode. |
| 124 | lock: Whether to use file locking. |
| 125 | logger: Logger instance. |
| 126 | max_open_shards: Maximum open shards (0 = no limit). |
| 127 | """ |
| 128 | self.cache_dir = os.path.abspath(os.path.expanduser(cache_dir)) |
| 129 | self.num_shards = num_shards |
| 130 | self.map_size_per_shard = map_size_per_shard |
| 131 | self.readonly = readonly |
| 132 | self.lock = lock |
| 133 | self.logger = logger |
| 134 | # 0 means no limit, default allows all shards open |
| 135 | self.max_open_shards = max_open_shards if max_open_shards > 0 else num_shards |
| 136 | |
| 137 | # Use OrderedDict for LRU-style shard management |
| 138 | self._envs: OrderedDict[int, lmdb.Environment] = OrderedDict() |
| 139 | self._lock = threading.Lock() |
| 140 | os.makedirs(self.cache_dir, exist_ok=True) |
| 141 | |
| 142 | if self.logger: |
| 143 | self.logger.info( |
| 144 | f"ShardedLMDBManager initialized: cache_dir={self.cache_dir}, " |
| 145 | f"num_shards={num_shards}, map_size_per_shard={map_size_per_shard/1024**3:.1f}GB" |
| 146 | ) |
| 147 | |
| 148 | @classmethod |
| 149 | def get_instance( |