482 lines
19 KiB
Plaintext
482 lines
19 KiB
Plaintext
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<Address, Connection>
|
|
pending_operations: Map<OpID, Operation>
|
|
|
|
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<ChunkTask>
|
|
bytes_transferred: Integer
|
|
result_data: Any
|
|
|
|
class ChunkTask:
|
|
# Used for tracking individual chunk uploads/downloads
|
|
chunk_index: Integer
|
|
hash: String
|
|
target_servers: List<Address>
|
|
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)
|
|
} |