note that: MS = metadata server CS = chunk server CLI = client ========================================================= === METADATA SERVER PROCESS ============================= ========================================================= metadata_server_init: Start a TCP server at the interface and port specified using options --addr and --port (default 127.0.0.1 and 8080) Initialize the file tree as empty Initialize the WAL and, if it exists already, replay its contents into the file tree metadata_server_free: Free resources associated to the MS metadata_server_tick: for each network event: if connection established: mark the connection as UNKNOWN if disconnect: if the connection was marked as CS: Free the information associated to it if message received: if the connection is marked as UNKNOWN: if the message is from a CS: mark the connection as CS and associated to it a struct for the information associated to it. else mark the connection as CLI if the peer is a client: if the message has type CREATE: Read path_len, path, and is_dir from the message If is_dir is false: Read chunk_size from message Else: Set chunk_size to 0 Append the create operation to the WAL Attempt to create entity in the file_tree If file_tree creation fails: Send CREATE_ERROR with the error description Else: Send CREATE_SUCCESS containing the new generation number if the message has type DELETE: Read expect_gen, path_len, and path from message Append delete operation to the WAL Attempt to delete entity in the file_tree If file_tree deletion fails: Send DELETE_ERROR with the error description Else: Send DELETE_SUCCESS (no payload) if the message has type LIST: Read expect_gen, path_len, and path from message Attempt to list directory contents from file_tree If listing fails: Send LIST_ERROR with error description Else: Send LIST_SUCCESS containing: - Directory generation - Truncated flag (if results exceed MAX_LIST_SIZE) - Item count - List of items (generation, is_dir, name_len, name) if the message has type READ: Read expect_gen, path_len, path, offset, and length Attempt to read file info from file_tree If read fails: Send READ_ERROR containing: - Error description - A list of suggested chunk server addresses for writing (load balanced) NOTE: Providing write locations on read error allows clients to handle "missing file" scenarios by immediately writing if desired. Else: Send READ_SUCCESS containing: - File generation - Chunk size - Actual bytes available (may be less than requested length) - Number of chunks - List of chunks, where each contains: - SHA256 Hash - List of chunk servers holding this chunk (holders) - A list of suggested chunk server addresses for writing new chunks (load balanced via choose_servers_for_write) if the message has type WRITE: Read expect_gen, flags, path_len, path Read offset, length, num_chunks For each chunk in num_chunks: Read the hash and the list of Chunk Server addresses (locations) that hold it (This implies the client has already uploaded the data to these servers) Append write operation to WAL Attempt to apply write to file_tree (returns new_gen and a list of removed_hashes) If result is NOENT (File Not Found) AND flags has TOASTY_WRITE_CREATE_IF_MISSING: - Append create operation (default chunk_size 4096) to WAL - Create entity in file_tree - Retry the write operation to file_tree If write fails: Send WRITE_ERROR with error description Else: For each new chunk hash provided in the message: Find the chunk servers associated with that hash in the message Add the hash to those servers' ms_add_list For each removed_hash returned by the file_tree: Find all chunk servers currently holding this hash Add the hash to their ms_rem_list (marking it for deletion) Send WRITE_SUCCESS containing the new generation number if the peer is a CS: if the message has type AUTH: Add all IPv4/port and IPv6/port pairs to the data associated to this connection. Then, respond with a AUTH_RESPONSE message. if the message has type SYNC: For each hash received: If it's not in use by the file tree: ignore it If less than REPLICATION_FACTOR CS are holding it: Add this hash to this CS's add_list Respond with a SYNC_2 message containing the add_list and the rem_list. if the message has type SYNC_3: Create an empty tmp_list For each hash received: We expect the hash not to be in the add_list or the old_list. Only hashes that were actually expected to be in on the server should be recovered. Add the hash to tmp_list. For each CS holding this hash in their old_list or add_list: Write its address to the output message Merge add_list into old_list Remove elements in tmp_list from old_list Set old_list equal to tmp_list Send the message as SYNC_4 For each chunk server: If its last response was more than RESPONSE_TIME_LIMIT seconds ago: Drop the connection Set the current timeout to the next most imminent event ========================================================= === CHUNK SERVER PROCESS ================================ ========================================================= chunk_server_init: Start a TCP server at the interface and port specified using options --addr and --port (default 127.0.0.1 and 8081) Create a directory for chunk files at path specified using option --path (default "chunk_server_data") Connect to MS using the remote address and port specified using options --remote-addr and --remote-port (default 127.0.0.1, 8080) and send an AUTH message containing the list of local IPv4/port and IPv6/port pairs. chunk_server_free: free all resources start_download: if not downloading already: get a descriptor from the download_targets array and send DOWNLOAD_CHUNK message with the specified hash. chunk_server_tick: for each network event: if connection established: do nothing if disconnect: if the connection was to MS: disconnect_time = current_time if the connection was to a CS: downloading = false if message received: if from MS: if tagged as AUTH_RESPONSE: Read all chunk file names from the data directory and convert them into hashes by decoding them as hexadecimal. Add all of these hashes to the add_list. No message is sent back to MS in response. if tagged as SYNC_2: For each hash in the first list received by MS: If no chunk file is associated to it: add it to a tmp_list else: remove it from the rem_list and from the add_list. For each hash still in the add_list: Add it to the rem_list Mark the add_list as empty For each hash in the second list received by MS: Add it to rem_list Send SYNC_3 message with a list of hashes made by those in tmp_list and lst_list. if tagged as SYNC_4: At this point MS knows which chunks this CS is missing, so it sent back information for where to download them. For each element in the list: Add a download descrptor to download_targets list. start_download() (other cases are not allowed) if from CS: We may get messages from other CS when trying to download a chunk from it that was lost locally. if tagged DOWNLOAD_CHUNK_ERROR: Start the next download if tagged DOWNLOAD_CHUNK_SUCCESS: Create a chunk file with the received data, then add its hash to the add_list. start_download() if from CLI: if tagged CREATE_CHUNK: Take the received data and make a chunk file for it, then add its hash to the add_list. Respond with a CREATE_CHUNK_SUCCESS message. if tagged UPLOAD_CHUNK: Load the chunk referred by the client and patch it with the data it sent at the specified offset, then calculate its hash and store it as a new chunk. Add its hash to the add_list. Respond with a UPLOAD_CHUNK_SUCCESS message. if tagged DOWNLOAD_CHUNK: Load the chunk referred by the client and respond with its data in a DOWNLOAD_CHUNK_SUCCESS message. For each expired item in rem_list: Delete its associated chunk file If the connection to MS is intact and more than SYNC_INTERVAL seconds elapsed since the last sync: Send a SYNC message containing all add_list items. start_download() If there is no connection to MS and more than RECONNECT_DELAY seconds passed: Connect to the metadata server and send a message of type AUTH containing the list of our IPv4s/port pairs and IPv6/port pairs. Set the current timeout to the earliest time, considering the next reconnect attempt timeto MS if disconnected and the most imminent rem_list deadline. ========================================================= === CLIENT LIBRARY ====================================== ========================================================= # ------------------------------------------------------------------------------ # 1. DATA STRUCTURES # ------------------------------------------------------------------------------ class ToastyClient: metadata_server_conn: Connection chunk_server_pool: Map pending_operations: Map class Operation: id: Integer type: Enum(READ, WRITE, CREATE, DELETE, LIST) status: Enum(PENDING, COMPLETED, FAILED) # Context path: String user_buffer: Byte[] offset: Integer length: Integer flags: Bitmask expected_generation: Integer # Internal State chunks_to_process: List bytes_transferred: Integer result_data: Any class ChunkTask: # Used for tracking individual chunk uploads/downloads chunk_index: Integer hash: String target_servers: List
status: Enum(WAITING, IN_PROGRESS, DONE) final_hash: String # Returned by CS after upload/patch # ------------------------------------------------------------------------------ # 2. PUBLIC BLOCKING API (User Facing) # ------------------------------------------------------------------------------ function toasty_connect(address, port): client = new ToastyClient() client.metadata_server_conn = tcp_connect(address, port) return client function toasty_read(client, path, offset, length): op_handle = begin_read_op(client, path, offset, length) return wait_for_result(client, op_handle) function toasty_write(client, path, data, offset, flags): op_handle = begin_write_op(client, path, data, offset, flags) return wait_for_result(client, op_handle) # (Create, Delete, and List follow the same pattern: begin_op -> wait_for_result) # ------------------------------------------------------------------------------ # 3. INTERNAL ASYNCHRONOUS LOGIC # ------------------------------------------------------------------------------ # --- READ LOGIC --- function begin_read_op(client, path, offset, length): op = new Operation(type=READ, path, offset, length) # Step 1: Ask Metadata Server (MS) for the file layout send_message(client.metadata_server_conn, type=READ_REQUEST, path=path, range=(offset, length)) return register_operation(client, op) function handle_read_metadata_response(op, msg): if msg.type == ERROR: op.status = FAILED return # Step 2: Calculate chunk plan chunk_size = msg.chunk_size op.chunks_to_process = calculate_required_chunks(op.offset, op.length, chunk_size) # Step 3: Queue downloads from Chunk Servers (CS) for chunk in op.chunks_to_process: # MS returns a list of holders for each chunk hash best_cs = load_balance_select(chunk.holders) conn = get_cs_connection(best_cs) send_message(conn, type=DOWNLOAD_CHUNK, hash=chunk.hash) function handle_chunk_download_success(op, msg): # Step 4: Assemble data copy_bytes(src=msg.data, dest=op.user_buffer, at_offset=msg.chunk_index) mark_task_complete(op, msg.chunk_index) if all_tasks_complete(op): op.status = COMPLETED # --- WRITE LOGIC (The Complex Part) --- function begin_write_op(client, path, data, offset, flags): op = new Operation(type=WRITE, path, data, offset, flags) # Step 1: Send READ to MS first. # We need the current generation number and existing chunk hashes to perform # patches or ensure consistency, even if we are overwriting. send_message(client.metadata_server_conn, type=READ_REQUEST, path=path, range=(offset, len(data))) return register_operation(client, op) function handle_write_metadata_ready(op, msg): # Step 2: Prepare Upload Tasks # If file missing & CREATE_IF_MISSING flag is set, we use defaults (size 0, gen 0). current_generation = msg.generation chunk_size = msg.chunk_size write_targets = msg.write_locations # Recommended CS IPs from MS # Slice user data into chunks sliced_chunks = split_into_chunks(op.user_buffer, chunk_size) for slice in sliced_chunks: task = new ChunkTask() task.target_servers = write_targets # We must replicate to 3 servers # Determine if this is a Create (New) or Update (Patch) if slice.index >= len(msg.existing_hashes): task.type = NEW_CHUNK else: task.type = PATCH_CHUNK task.base_hash = msg.existing_hashes[slice.index] op.chunks_to_process.add(task) # Step 3: Start Uploads (Parallel up to limit) pump_upload_queue(op) function pump_upload_queue(op): while active_uploads(op) < MAX_PARALLEL_UPLOADS: task = get_next_waiting_task(op) if not task: break # We must upload to N replicas (usually 3) for replica_addr in task.target_servers: conn = get_cs_connection(replica_addr) if task.type == NEW_CHUNK: # Full upload send_message(conn, type=CREATE_CHUNK, data=task.data) else: # Differential patch (Rsync-style or offset-based) send_message(conn, type=UPLOAD_CHUNK, base_hash=task.base_hash, patch_data=task.data) function handle_upload_ack(op, msg): # Step 4: Collect new hashes task = get_task(op, msg.task_id) # The CS computes the new SHA256 after applying the patch/data task.final_hash = msg.new_hash task.replicas_confirmed++ if task.replicas_confirmed >= REQUIRED_REPLICATION: task.status = DONE if all_tasks_complete(op): commit_write(op) else: pump_upload_queue(op) # Start next batch function commit_write(op): # Step 5: Atomic Commit at Metadata Server commit_msg = new Message(type=WRITE_REQUEST) commit_msg.path = op.path commit_msg.expected_generation = op.expected_generation # Optimistic Locking commit_msg.flags = op.flags # Map the linear chunk index to the new Hash + Location list commit_msg.chunks = map_chunks_to_hashes_and_locations(op.chunks_to_process) send_message(client.metadata_server_conn, commit_msg) function handle_commit_response(op, msg): if msg.type == WRITE_SUCCESS: op.status = COMPLETED op.result_data = msg.new_generation elif msg.type == ERROR and msg.code == GENERATION_MISMATCH: # Optimistic concurrency failed (someone else wrote). # Retry the whole operation from Step 1 (READ). reset_and_retry(op) else: op.status = FAILED # ------------------------------------------------------------------------------ # 4. EVENT LOOP (Driver) # ------------------------------------------------------------------------------ function wait_for_result(client, op_handle): while get_op_status(op_handle) == PENDING: # Wait for network activity (Poll/Select) events = wait_for_network_events(timeout=DEFAULT_TIMEOUT) for event in events: if event.type == DISCONNECT: handle_disconnect(client, event.conn) continue # Route message to the specific Operation ID embedded in the packet tag op = client.pending_operations[event.tag.op_id] msg = read_message(event.conn) # Dispatcher switch (op.type, msg.type): case (READ, READ_SUCCESS): handle_read_metadata_response(op, msg) case (READ, CHUNK_DATA): handle_chunk_download_success(op, msg) case (WRITE, READ_SUCCESS): handle_write_metadata_ready(op, msg) case (WRITE, UPLOAD_ACK): handle_upload_ack(op, msg) case (WRITE, WRITE_SUCCESS): handle_commit_response(op, msg) case (_, ERROR): handle_error(op, msg) return get_op_result(op_handle) }