import 'dart:convert'; import 'dart:io'; import 'package:sweepstore/structures.dart'; import 'helpers.dart'; int roundToNearest16(int value) { int rounded = (value + 15) & ~15; return rounded; } void initialiseSweepstoreHeader(RandomAccessFile file, { int concurrentWorkers = 4, }) { SweepstoreHeaderWriter header = SweepstoreHeaderWriter(file); if (header.magicNumber == 'SWPT') { throw ArgumentError('Sweepstore file is already initialised.'); } SweepstoreConcurrencyHeaderWriter concurrencyHeader = SweepstoreConcurrencyHeaderWriter(header); header.magicNumber = 'SWPT'; header.version = "undefined"; header.addressTablePointer = SweepstorePointer.nullptr; header.freeListCount = 0; header.isFreeListLifted = false; concurrencyHeader.masterIdentifier = 0; concurrencyHeader.masterHeartbeat = 0; concurrencyHeader.numberOfWorkers = concurrentWorkers; concurrencyHeader.isReadAllowed = false; for (int i = 0; i < concurrentWorkers; i++) { SweepstoreWorkerTicket ticket = concurrencyHeader[i]; ticket.write( identifier: 0, workerHeartbeat: 0, ticketState: SweepstoreTicketState.FREE, ticketOperation: SweepstoreTicketOperation.NONE, keyHash: 0, writePointer: SweepstorePointer.nullptr, writeSize: 0, ); } } class SweepstoreHeader { final RandomAccessFile _file; SweepstoreHeader(this._file); // Offset 0 - 4 bytes String get magicNumber { _file.setPositionSync(0); final bytes = _file.readSync(4); return String.fromCharCodes(bytes); } // Offset 4 - 12 bytes String get version { _file.setPositionSync(4); String version = utf8.decode(_file.readSync(12)).trim(); return version; } // Offset 16 - 8 bytes SweepstorePointer get addressTablePointer { _file.setPositionSync(16); final address = _file.readIntSync(8); return SweepstorePointer(address); } // Offset 24 - 4 bytes int get freeListCount { _file.setPositionSync(24); int count = _file.readIntSync(4); return count; } // Offset 28 - 1 byte bool get isFreeListLifted { _file.setPositionSync(28); int flag = _file.readIntSync(1); return flag != 0; } SweepstoreConcurrencyHeader get concurrency => SweepstoreConcurrencyHeader(this); } class SweepstoreHeaderWriter extends SweepstoreHeader { SweepstoreHeaderWriter(RandomAccessFile file) : super(file); // Offset 0 - 4 bytes void set magicNumber(String value) { if (value.length != 4) { throw ArgumentError('Magic number must be exactly 4 characters long'); } _file.setPositionSync(0); _file.writeFromSync(value.codeUnits); } // Offset 4 - 12 bytes void set version(String value) { if (value.length > 11) { throw ArgumentError('Version string must be at most 11 characters long'); } _file.setPositionSync(4); _file.writeFromSync(utf8.encode(" " + value.padRight(11, ' ').substring(0, 11))); } // Offset 16 - 8 bytes void set addressTablePointer(SweepstorePointer pointer) { _file.setPositionSync(16); _file.writeIntSync(pointer.address, 8); } // Offset 24 - 4 bytes void set freeListCount(int value) { _file.setPositionSync(24); _file.writeIntSync(value, 4); } // Offset 28 - 1 byte void set isFreeListLifted(bool lifted) { _file.setPositionSync(28); _file.writeIntSync(lifted ? 1 : 0, 1); } } class SweepstoreConcurrencyHeader { final SweepstoreHeader _header; SweepstoreConcurrencyHeader(this._header); // Offset 29 - 8 bytes int get masterIdentifier { _header._file.setPositionSync(29); int id = _header._file.readIntSync(8); return id; } // Offset 37 - 4 bytes int get masterHeartbeat { _header._file.setPositionSync(37); int heartbeat = _header._file.readIntSync(4); return heartbeat; } // Offset 41 - 4 bytes int get numberOfWorkers { _header._file.setPositionSync(41); int numWorkers = _header._file.readIntSync(4); return numWorkers; } // Offset 45 - 1 byte bool get isReadAllowed { _header._file.setPositionSync(45); int flag = _header._file.readIntSync(1); return flag != 0; } SweepstoreWorkerTicket operator [](int ticketIndex) { if (ticketIndex < 0 || ticketIndex >= numberOfWorkers) { throw RangeError.index(ticketIndex, this, 'ticketIndex', null, numberOfWorkers); } return SweepstoreWorkerTicket(ticketIndex, this); } } class SweepstoreConcurrencyHeaderWriter extends SweepstoreConcurrencyHeader { SweepstoreConcurrencyHeaderWriter(SweepstoreHeader header) : super(header); // Offset 29 - 8 bytes void set masterIdentifier(int id) { _header._file.setPositionSync(29); _header._file.writeIntSync(id, 8); } // Offset 37 - 4 bytes void set masterHeartbeat(int heartbeat) { _header._file.setPositionSync(37); _header._file.writeIntSync(heartbeat, 4); } // Offset 41 - 4 bytes void set numberOfWorkers(int numWorkers) { _header._file.setPositionSync(41); _header._file.writeIntSync(numWorkers, 4); } // Offset 45 - 1 byte void set isReadAllowed(bool allowed) { _header._file.setPositionSync(45); _header._file.writeIntSync(allowed ? 1 : 0, 1); } } final int endOfStaticHeaderOffset = roundToNearest16(46); class SweepstoreWorkerTicket { static final int ticketSize = roundToNearest16(29); final SweepstoreConcurrencyHeader _concurrencyHeader; final int ticketIndex; SweepstoreWorkerTicket(this.ticketIndex, this._concurrencyHeader); // All offsets are relative to the start of the workers ticket int get _baseOffset => endOfStaticHeaderOffset + (ticketIndex * ticketSize); // // Offset 0 - 4 bytes // int get identifier { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset, _baseOffset + 4); // _concurrencyHeader._header._file.setPositionSync(_baseOffset); // int id = _concurrencyHeader._header._file.readIntSync(4); // _concurrencyHeader._header._file.unlockSync(_baseOffset, _baseOffset + 4); // return id; // } // // // Offset 4 - 4 bytes // int get workerHeartbeat { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset + 4, _baseOffset + 8); // _concurrencyHeader._header._file.setPositionSync(_baseOffset + 4); // int heartbeat = _concurrencyHeader._header._file.readIntSync(4); // _concurrencyHeader._header._file.unlockSync(_baseOffset + 4, _baseOffset + 8); // return heartbeat; // } // // // Offset 8 - 1 byte // SweepstoreTicketState get ticketState { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset + 8, _baseOffset + 9); // _concurrencyHeader._header._file.setPositionSync(_baseOffset + 8); // int stateValue = _concurrencyHeader._header._file.readIntSync(1); // _concurrencyHeader._header._file.unlockSync(_baseOffset + 8, _baseOffset + 9); // return SweepstoreTicketState.values[stateValue]; // } // // // Offset 9 - 1 byte // SweepstoreTicketOperation get ticketOperation { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset + 9, _baseOffset + 10); // _concurrencyHeader._header._file.setPositionSync(_baseOffset + 9); // int operationValue = _concurrencyHeader._header._file.readIntSync(1); // _concurrencyHeader._header._file.unlockSync(_baseOffset + 9, _baseOffset + 10); // return SweepstoreTicketOperation.values[operationValue]; // } // // // Offset 10 - 8 bytes // int get keyHash { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset + 10, _baseOffset + 18); // _concurrencyHeader._header._file.setPositionSync(_baseOffset + 10); // int hash = _concurrencyHeader._header._file.readIntSync(8); // _concurrencyHeader._header._file.unlockSync(_baseOffset + 10, _baseOffset + 18); // return hash; // } // // // Offset 18 - 8 bytes // SweepstorePointer get writePointer { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset + 18, _baseOffset + 26); // _concurrencyHeader._header._file.setPositionSync(_baseOffset + 18); // int address = _concurrencyHeader._header._file.readIntSync(8); // _concurrencyHeader._header._file.unlockSync(_baseOffset + 18, _baseOffset + 26); // return SweepstorePointer(address); // } // // // Offset 26 - 4 bytes // int get writeSize { // _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset + 26, _baseOffset + 30); // _concurrencyHeader._header._file.setPositionSync(_baseOffset + 26); // int size = _concurrencyHeader._header._file.readIntSync(4); // _concurrencyHeader._header._file.unlockSync(_baseOffset + 26, _baseOffset + 30); // return size; // } // Writer void write({ int? identifier, int? workerHeartbeat, SweepstoreTicketState? ticketState, SweepstoreTicketOperation? ticketOperation, int? keyHash, SweepstorePointer? writePointer, int? writeSize, }) { try { _concurrencyHeader._header._file.lockSync(FileLock.blockingExclusive, _baseOffset, _baseOffset + ticketSize); _concurrencyHeader._header._file.setPositionSync(_baseOffset); List existingBuffer = _concurrencyHeader._header._file.readSync(ticketSize); RandomAccessMemory buffer = RandomAccessMemory(existingBuffer); if (identifier != null) { buffer.setPositionSync(0); buffer.writeIntSync(identifier, 4); } if (workerHeartbeat != null) { buffer.setPositionSync(4); buffer.writeIntSync(workerHeartbeat, 4); } if (ticketState != null) { buffer.setPositionSync(8); buffer.writeIntSync(ticketState.index, 1); } if (ticketOperation != null) { buffer.setPositionSync(9); buffer.writeIntSync(ticketOperation.index, 1); } if (keyHash != null) { buffer.setPositionSync(10); buffer.writeIntSync(keyHash, 8); } if (writePointer != null) { buffer.setPositionSync(18); buffer.writeIntSync(writePointer.address, 8); } if (writeSize != null) { buffer.setPositionSync(26); buffer.writeIntSync(writeSize, 4); } // Pad the rest of the ticket with zeros if necessary buffer.setPositionSync(30); while (buffer.positionSync() < ticketSize) { buffer.writeIntSync(0, 1); } _concurrencyHeader._header._file.setPositionSync(_baseOffset); _concurrencyHeader._header._file.writeFromSync(buffer.toUint8List()); _concurrencyHeader._header._file.flushSync(); } finally { _concurrencyHeader._header._file.unlockSync(_baseOffset, _baseOffset + ticketSize); } } bool writable() { try { _concurrencyHeader._header._file.lockSync( FileLock.blockingExclusive, _baseOffset, _baseOffset + ticketSize ); // Successfully locked - immediately unlock and return true _concurrencyHeader._header._file.unlockSync(_baseOffset, _baseOffset + ticketSize); return true; } catch (e) { // Lock failed - already held by another process return false; } } SweepstoreWorkerTicketSnapshot snapshot() { _concurrencyHeader._header._file.lockSync(FileLock.blockingShared, _baseOffset, _baseOffset + ticketSize); _concurrencyHeader._header._file.setPositionSync(_baseOffset); List existingBuffer = _concurrencyHeader._header._file.readSync(ticketSize); RandomAccessMemory buffer = RandomAccessMemory(existingBuffer); _concurrencyHeader._header._file.unlockSync(_baseOffset, _baseOffset + ticketSize); buffer.setPositionSync(0); int identifier = buffer.readIntSync(4); int workerHeartbeat = buffer.readIntSync(4); SweepstoreTicketState ticketState = SweepstoreTicketState.values[buffer.readIntSync(1)]; SweepstoreTicketOperation ticketOperation = SweepstoreTicketOperation.values[buffer.readIntSync(1)]; int keyHash = buffer.readIntSync(8); SweepstorePointer writePointer = SweepstorePointer(buffer.readIntSync(8)); int writeSize = buffer.readIntSync(4); return SweepstoreWorkerTicketSnapshot._( ticketIndex: ticketIndex, identifier: identifier, workerHeartbeat: workerHeartbeat, ticketState: ticketState, ticketOperation: ticketOperation, keyHash: keyHash, writePointer: writePointer, writeSize: writeSize, ); } } class SweepstoreWorkerTicketSnapshot { final int ticketIndex; final int identifier; final int workerHeartbeat; final SweepstoreTicketState ticketState; final SweepstoreTicketOperation ticketOperation; final int keyHash; final SweepstorePointer writePointer; final int writeSize; SweepstoreWorkerTicketSnapshot._({ required this.ticketIndex, required this.identifier, required this.workerHeartbeat, required this.ticketState, required this.ticketOperation, required this.keyHash, required this.writePointer, required this.writeSize, }); }