diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..7c75d96 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +# Ensure shell scripts always use LF line endings +*.sh text eol=lf diff --git a/.gitignore b/.gitignore index 2b915ce..7eb8ffc 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,8 @@ *.o *.a chunk_server_data_* +# Coverage reports +coverage_report/ +*.gcov +*.gcda +*.gcno diff --git a/DESIGN.txt b/DESIGN.txt index 7f4447d..aefba33 100644 --- a/DESIGN.txt +++ b/DESIGN.txt @@ -137,37 +137,75 @@ Chunk Management: servers need to forget some copies. If they are under-replicated, some chunk servers need to copy chunks from elsewhere. - Event 1: A chunk server connects - If the chunk server is holding some chunks, for each chunk - one of the following must be true: - - The chunk is not used by the file system - => Mark the chunk for removal - - The chunk is used and, with this copy, perfectly replicated - => Do nothing - - The chunk is used and under-replicated - => ??? - - The chunk is used and over-replicated - => Mark the chunk for removal + Metadata server variables for a chunk server: + ms_old_list: List of chunks that are known to be held by CS + ms_add_list: List of chunks that should be held by CS + ms_rem_list: List of chunks that may be held by CS but should removed from it - Event 2: A write operation on metadata (adding chunks) - TODO + Chunk server variables: + cs_add_list: List of chunks added since the last update + cs_rem_list: List of chunks marked for removal after a timeout + cs_lst_list: List of chunks that were lost due to errors or forceful removals of chunk files - Event 3: A chunk server disconnects - A number of chunks are lost, but there is a chance the server will come back - in a short while (10s or so). + Metadata change for write: + When clients commit a write by adding new hashes to the metadata, + MS adds those hashes to the ms_add_list for the CS where the client + uploaded the chunks. If a hash was overwritten and became useless, + that hash is added to the ms_rem_list for all CS holding it. - TODO + Metadata change for delete: + All hashes that are no longer reachable by the file tree are added + to the ms_rem_list of their holders - Event 4: A write operation on metadata occurs (overwriting old chunks) - Assuming the write overwrites some chunks: - - The chunks may still be referenced by some other file or section of the same file - => Do nothing - - The chunks are not used anymore - => Find all chunk servers holding the chunk and mark them for removal + Chunk upload: + When a chunk is uploaded to a chunk server, its hash is added to + the cs_add_list. - Event 5: A delete operation on metadata occurs - Do the same as event 4 + Periodically on the chunk server: + Elements in the cs_rem_list have a 30 minute timeout after which they are deleted + permanently. - Event 6: A chunk is corrupted or removed forcefully from the chunk server - The chunk server adds the chunk name to a list of lost chunks and - sends them to the metadata server at the next periodic update. + Periodically: + CS sends cs_add_list to MS + MS may add a subset of cs_add_list to ms_add_list based on the chunk replication and distribution policy + MS sends ms_add_list and ms_rem_list to CS + CS (1) Adds all elements from ms_rem_list to cs_rem_list + (2) Elements in ms_add_list that are not held by the chunk server are added to + a temporary list tmp_list + (3) Removes elements in ms_add_list from cs_add_list and cs_rem_list, then merges cs_add_list into cs_rem_list and clears cs_add_list. + (4) Elements in cs_lst_list are added to tmp_list, then cs_lst_list is cleared. + (6) tmp_list is sent to MS + (7) cs_add_list is cleared + MS (1) Receives tmp_list and sends download locations to CS for those chunks + (2) Merges ms_add_list into ms_old_list, then removes all items in tmp_list from ms_old_list + (3) Sets ms_add_list equal to tmp_list + + Chunk replication and distribution policy: + During an update, when CS reports a new chunk to MS, MS has to decide whether + to allow the CS to keep it or not. + + There are 4 cases: + - The chunk is useless (not referenced by any file) + - The chunk is under-replicated even counting the new copy + - The chunk is properly replicated with the new copy + - The chunk is over-replicated with the new copy + + If the chunk is not referenced by the file tree, do nothing. + + If the chunk is properly replicated or under-replicated, add it to the ms_add_list. + + If the chunk is over-replicated, either don't add it to the ms_add_list or add it to the ms_rem_list of some other holder. + + TODO: The way chunk servers tell about the chunks they are holding + recently changed. Now instead of sending a full chunk list when + connecting, they send batches of chunks to the metadata server + during state updates. It also changed that now chunk servers + initiate updates. + + When a chunk server connects (and authenticates) + it sends alongside the auth message the hash of + the hash list of all chunks. The metadata server + then replies with a message saying whether it + already cached that chunk list or not. If it didn't, the chunk server sends the entire list + in chunks during state updates, with an increased + update frequency. diff --git a/Makefile b/Makefile index d8c62fc..ee34697 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,7 @@ -CFLAGS = -Wall -Wextra -ggdb +CFLAGS = -Wall -Wextra -ggdb -fsanitize=address,undefined +COVERAGE_CFLAGS = $(CFLAGS) --coverage +COVERAGE_LFLAGS = --coverage ifeq ($(OS),Windows_NT) LFLAGS = -lws2_32 @@ -13,16 +15,27 @@ CFILES = $(shell find src -name '*.c') HFILES = $(shell find src -name '*.h') OFILES = $(CFILES:.c=.o) -.PHONY: all clean +.PHONY: all clean coverage coverage-report coverage-html all: mousefs$(EXT) mousefs_random_test$(EXT) example_client$(EXT) libmousefs.a +coverage: mousefs_random_test_coverage$(EXT) + +coverage-report: + @./scripts/measure_coverage.sh 60 + +coverage-html: + @./scripts/measure_coverage.sh 60 --html + mousefs$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER mousefs_random_test$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_TEST +mousefs_random_test_coverage$(EXT): $(CFILES) $(HFILES) + gcc -o $@ $(CFILES) $(COVERAGE_CFLAGS) $(LFLAGS) $(COVERAGE_LFLAGS) -Iinc -DBUILD_TEST + example_client$(EXT): libmousefs.a gcc -o $@ examples/main.c $(CFLAGS) -lmousefs $(LFLAGS) -Iinc -L. @@ -33,12 +46,18 @@ libmousefs.a: $(OFILES) ar rcs $@ $^ clean: - rm -f \ - mousefs.exe \ - mousefs.out \ - mousefs_random_test.exe \ - mousefs_random_test.out \ - example_client.exe \ - example_client.out \ - libmousefs.a \ - src/*.o + rm -f \ + mousefs.exe \ + mousefs.out \ + mousefs_random_test.exe \ + mousefs_random_test.out \ + mousefs_random_test_coverage.exe \ + mousefs_random_test_coverage.out \ + example_client.exe \ + example_client.out \ + libmousefs.a \ + src/*.o \ + src/*.gcda \ + src/*.gcno \ + *.gcda \ + *.gcno diff --git a/TODO.txt b/TODO.txt index 8061155..e495686 100644 --- a/TODO.txt +++ b/TODO.txt @@ -8,3 +8,11 @@ - find a way to remove over-replicated chunks and formalize the chunk management policy - add logging of: - when chunk servers connect and disconnect to the metadata server +- Should list scenarios that need testing, like those where chunks would be dropped +- Update DESIGN.txt and the code to remove the chunk list message. The information of chunks + held by chunk servers is now transmitted to the metadata server during state updates + +Roadmap: + [ ] Complete all endpoints + [ ] Add fault injections to the simulation + [ ] Implement WAL diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..bcaac89 --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,48 @@ +# Branch Coverage Measurement + +This directory contains scripts for measuring branch coverage of the random simulation. + +## Prerequisites + +The HTML report generation requires `lcov`: + +```bash +sudo apt-get install lcov +``` + +## Usage + +### Quick text summary (5 second simulation) +```bash +make coverage-report +``` + +### HTML report with branch details (5 second simulation) +```bash +make coverage-html +``` + +### Custom duration +```bash +./scripts/measure_coverage.sh 10 # 10 second run, text output +./scripts/measure_coverage.sh 10 --html # 10 second run, HTML output +``` + +## What gets measured + +The coverage tool: +- Builds the test binary with coverage instrumentation (`--coverage` flag) +- Runs the random simulation for the specified duration +- Generates reports showing which branches were executed + +## Output + +**Text report**: Shows per-file and total branch coverage percentages + +**HTML report**: Interactive report generated by lcov/genhtml showing: +- Overall coverage summary +- Per-file coverage breakdown +- Source code with execution counts +- Branch coverage details + +The HTML report is generated in `coverage_report/index.html` diff --git a/scripts/fix_line_endings.sh b/scripts/fix_line_endings.sh new file mode 100755 index 0000000..d2af20f --- /dev/null +++ b/scripts/fix_line_endings.sh @@ -0,0 +1,18 @@ +#!/bin/bash +# Fix line endings for coverage scripts + +echo "Fixing line endings for coverage scripts..." + +# Convert CRLF to LF +dos2unix scripts/measure_coverage.sh 2>/dev/null || sed -i 's/\r$//' scripts/measure_coverage.sh +dos2unix scripts/generate_coverage_html.sh 2>/dev/null || sed -i 's/\r$//' scripts/generate_coverage_html.sh + +# Ensure execute permissions +chmod +x scripts/measure_coverage.sh +chmod +x scripts/generate_coverage_html.sh + +echo "Done! Line endings fixed and execute permissions set." +echo "" +echo "You can now run:" +echo " make coverage-report" +echo " make coverage-html" diff --git a/scripts/generate_coverage_html.sh b/scripts/generate_coverage_html.sh new file mode 100755 index 0000000..c29788b --- /dev/null +++ b/scripts/generate_coverage_html.sh @@ -0,0 +1,31 @@ +#!/bin/bash +# Generate HTML coverage report using lcov + +set -e + +OUTPUT_DIR="coverage_report" + +# Check if lcov is installed +if ! command -v lcov &> /dev/null; then + echo "Error: lcov is not installed" + echo "Please install it with: sudo apt-get install lcov" + exit 1 +fi + +# Capture coverage data with branch coverage enabled +echo "Capturing coverage data..." +lcov --capture --directory . --output-file coverage.info --rc lcov_branch_coverage=1 + +# Filter out system headers if any exist +echo "Filtering coverage data..." +lcov --remove coverage.info '/usr/*' --output-file coverage.info --ignore-errors unused --rc lcov_branch_coverage=1 || cp coverage.info coverage.info.bak + +# Generate HTML report +echo "Generating HTML report..." +genhtml coverage.info --output-directory "$OUTPUT_DIR" --branch-coverage --rc lcov_branch_coverage=1 + +# Clean up +rm -f coverage.info + +echo "HTML report generated in $OUTPUT_DIR/" +echo "Open $OUTPUT_DIR/index.html in your browser" diff --git a/scripts/measure_coverage.sh b/scripts/measure_coverage.sh new file mode 100755 index 0000000..876ea3e --- /dev/null +++ b/scripts/measure_coverage.sh @@ -0,0 +1,113 @@ +#!/bin/bash +# Script to measure branch coverage of the random simulation + +set -e + +# Colors for output +GREEN='\033[0;32m' +BLUE='\033[0;34m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +echo -e "${BLUE}========================================${NC}" +echo -e "${BLUE}Branch Coverage Measurement Tool${NC}" +echo -e "${BLUE}========================================${NC}" +echo + +# Clean previous coverage data +echo -e "${YELLOW}Cleaning previous coverage data...${NC}" +rm -f src/*.gcda src/*.gcno *.gcda *.gcno +make clean > /dev/null 2>&1 + +# Build with coverage +echo -e "${YELLOW}Building with coverage instrumentation...${NC}" +make coverage + +# Run the simulation for a limited time +echo -e "${YELLOW}Running simulation...${NC}" +SIMULATION_TIME=${1:-5} # Default to 5 seconds if not specified +echo "Running for ${SIMULATION_TIME} seconds..." + +# Run simulation in background and kill it after specified time +timeout ${SIMULATION_TIME}s ./mousefs_random_test_coverage.out || true + +# Generate coverage reports +echo +echo -e "${YELLOW}Generating coverage reports...${NC}" + +# Find all .gcda files and generate coverage reports +GCDA_FILES=$(find . -name "*.gcda") + +TOTAL_BRANCHES=0 +TAKEN_BRANCHES=0 + +# Generate gcov reports from the coverage data files +for gcda_file in $GCDA_FILES; do + # Extract the source file name from the gcda filename + # Files are named like: mousefs_random_test_coverage.out-basic.gcda + basename=$(basename "$gcda_file" .gcda) + source_name=${basename#*-} # Remove prefix up to and including '-' + + # Run gcov to generate the .gcov file + gcov -b "$gcda_file" > /dev/null 2>&1 || true +done + +# Parse gcov output to count branches +echo +echo -e "${BLUE}========================================${NC}" +echo -e "${BLUE}Branch Coverage Summary${NC}" +echo -e "${BLUE}========================================${NC}" +echo + +# Process each .gcov file +for gcov_file in *.c.gcov; do + if [ -f "$gcov_file" ]; then + # Extract branch statistics from the gcov file + branches=$(grep -c "^branch" "$gcov_file" 2>/dev/null) + if [ -z "$branches" ]; then branches=0; fi + + if [ "$branches" -gt 0 ]; then + taken_count=$(grep "^branch" "$gcov_file" 2>/dev/null | grep -c "taken [1-9]" 2>/dev/null || echo "0") + # Clean up the value - remove any whitespace/newlines + taken_count=$(echo "$taken_count" | tr -d '[:space:]') + if [ -z "$taken_count" ] || [ "$taken_count" = "" ]; then taken_count=0; fi + + TOTAL_BRANCHES=$((TOTAL_BRANCHES + branches)) + TAKEN_BRANCHES=$((TAKEN_BRANCHES + taken_count)) + percentage=$((taken_count * 100 / branches)) + filename=$(echo "$gcov_file" | sed 's/.gcov$//') + printf "%-40s %5d / %5d branches (%3d%%)\n" "$filename" "$taken_count" "$branches" "$percentage" + fi + fi +done + +echo +echo -e "${BLUE}========================================${NC}" +if [ "$TOTAL_BRANCHES" -gt 0 ]; then + COVERAGE_PERCENT=$((TAKEN_BRANCHES * 100 / TOTAL_BRANCHES)) + echo -e "${GREEN}Total: $TAKEN_BRANCHES / $TOTAL_BRANCHES branches reached (${COVERAGE_PERCENT}%)${NC}" +else + echo -e "${YELLOW}No branch coverage data found${NC}" +fi +echo -e "${BLUE}========================================${NC}" +echo + +# Generate HTML report if requested +if [ "$2" == "--html" ]; then + echo -e "${YELLOW}Generating HTML coverage report...${NC}" + # Get the directory where this script is located + SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + "$SCRIPT_DIR/generate_coverage_html.sh" + echo -e "${GREEN}HTML report generated in coverage_report/index.html${NC}" + echo "Open with: firefox coverage_report/index.html (or your preferred browser)" + echo +fi + +# Clean up gcov files unless HTML was requested +if [ "$2" != "--html" ] && [ "$2" != "--detailed" ]; then + echo -e "${YELLOW}Cleaning up coverage files...${NC}" + rm -f *.gcov +fi + +echo +echo -e "${GREEN}Coverage measurement complete!${NC}" diff --git a/src/basic.c b/src/basic.c index d6a72f0..9c81fd6 100644 --- a/src/basic.c +++ b/src/basic.c @@ -30,7 +30,7 @@ Time get_current_time(void) ok = sys_QueryPerformanceFrequency((LARGE_INTEGER*) &freq); if (!ok) return INVALID_TIME; - uint64_t res = 1000 * (double) count / freq; + uint64_t res = 1000000000 * (double) count / freq; return res; } #else @@ -45,12 +45,12 @@ Time get_current_time(void) uint64_t sec = time.tv_sec; if (sec > UINT64_MAX / 1000000000) return INVALID_TIME; - res = sec * 1000; + res = sec * 1000000000; uint64_t nsec = time.tv_nsec; if (res > UINT64_MAX - nsec) return INVALID_TIME; - res += nsec / 1000000; + res += nsec; return res; } diff --git a/src/chunk_server.c b/src/chunk_server.c index 6b48271..ecca8c7 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -4,10 +4,12 @@ #include #include "basic.h" +#include "byte_queue.h" #include "config.h" #include "sha256.h" #include "message.h" #include "file_system.h" +#include "tcp.h" #include "chunk_server.h" static void @@ -57,77 +59,6 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash) return 0; } -static void -removal_list_init(RemovalList *list) -{ - list->count = 0; - list->capacity = 0; - list->items = NULL; -} - -static void -removal_list_free(RemovalList *list) -{ - sys_free(list->items); -} - -static int -removal_list_find(RemovalList *list, SHA256 hash) -{ - for (int i = 0; i < list->count; i++) - if (!memcmp(&list->items[i].hash, &hash, sizeof(SHA256))) - return i; - return -1; -} - -static int -removal_list_add(RemovalList *list, SHA256 hash, Time marked_time) -{ - // Check if already in list - int idx = removal_list_find(list, hash); - if (idx >= 0) { - // Already marked, keep the original time - return 0; - } - - if (list->count == list->capacity) { - int new_capacity; - if (list->capacity == 0) - new_capacity = 8; - else - new_capacity = 2 * list->capacity; - - PendingRemoval *new_items = sys_malloc(new_capacity * sizeof(PendingRemoval)); - if (new_items == NULL) - return -1; - - if (list->capacity > 0) { - memcpy(new_items, list->items, list->count * sizeof(list->items[0])); - sys_free(list->items); - } - - list->items = new_items; - list->capacity = new_capacity; - } - - list->items[list->count++] = (PendingRemoval) { hash, marked_time }; - return 0; -} - -static void -removal_list_remove(RemovalList *list, SHA256 hash) -{ - int idx = removal_list_find(list, hash); - if (idx >= 0) { - // Remove by shifting remaining items - if (idx < list->count - 1) { - memmove(&list->items[idx], &list->items[idx + 1], - (list->count - idx - 1) * sizeof(list->items[0])); - } - list->count--; - } -} - static int chunk_store_init(ChunkStore *store, string path) { if (create_dir(path) && errno != EEXIST) @@ -221,6 +152,7 @@ static bool chunk_store_exists(ChunkStore *store, SHA256 hash) string path = hash2path(store, hash, buf); // Try to open the file to check if it exists + // TODO: this isn't right Handle fd; if (file_open(path, &fd) == 0) { file_close(fd); @@ -318,139 +250,99 @@ static void start_download_if_necessary(ChunkServer *state) } } -static int -process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView msg) +static int process_metadata_server_sync_2(ChunkServer *state, + int conn_idx, ByteView msg) { BinaryReader reader = { msg.ptr, msg.len, 0 }; - // Read header if (!binary_read(&reader, NULL, sizeof(MessageHeader))) - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); + return -1; uint32_t add_count; if (!binary_read(&reader, &add_count, sizeof(add_count))) - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); + return -1; + + Time current_time = get_current_time(); + if (current_time == INVALID_TIME) { + assert(0); // TODO + } + + HashSet tmp_list; + hash_set_init(&tmp_list); + + for (uint32_t i = 0; i < add_count; i++) { + + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) + return -1; + + // Elements in ms_add_list that are not held by the + // chunk server are added to a temporary list tmp_list + + if (!chunk_store_exists(&state->store, hash)) { + if (hash_set_insert(&tmp_list, hash) < 0) { + assert(0); // TODO + } + continue; + } + + timed_hash_set_remove(&state->cs_rem_list, hash); + hash_set_remove(&state->cs_add_list, hash); + } + + for (int i = 0; i < state->cs_add_list.count; i++) { + if (timed_hash_set_insert(&state->cs_rem_list, state->cs_add_list.items[i], current_time) < 0) { + assert(0); // TODO + } + } + + hash_set_clear(&state->cs_add_list); uint32_t rem_count; if (!binary_read(&reader, &rem_count, sizeof(rem_count))) - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - - SHA256 *add_list = sys_malloc(add_count * sizeof(SHA256)); - if (add_list == NULL) - return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); - - SHA256 *rem_list = sys_malloc(rem_count * sizeof(SHA256)); - if (rem_list == NULL) { - sys_free(add_list); - return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); - } - - for (uint32_t i = 0; i < add_count; i++) { - if (!binary_read(&reader, &add_list[i], sizeof(SHA256))) { - sys_free(add_list); - sys_free(rem_list); - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - } - } + return -1; for (uint32_t i = 0; i < rem_count; i++) { - if (!binary_read(&reader, &rem_list[i], sizeof(SHA256))) { - sys_free(add_list); - sys_free(rem_list); - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - } - } - if (binary_read(&reader, NULL, 1)) { - sys_free(add_list); - sys_free(rem_list); - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - } - - // Check that all items in the add_list are in the chunk directory - // Any hashes that are missing are added to a missing list - SHA256 *missing = NULL; - uint32_t num_missing = 0; - - Time current_time = get_current_time(); - - for (uint32_t i = 0; i < add_count; i++) { - // If chunk is in removal list, unmark it (remove from removal list) - removal_list_remove(&state->removal_list, add_list[i]); - - // Check if chunk exists in the chunk store - if (!chunk_store_exists(&state->store, add_list[i])) { - - // Chunk is missing, add to missing list - - if (missing == NULL) { - missing = sys_malloc(add_count * sizeof(SHA256)); - if (missing == NULL) { - assert(0); // TODO - } - } - missing[num_missing++] = add_list[i]; - } - } - - // Append items from the rem_list to the removal list with timestamps - for (uint32_t i = 0; i < rem_count; i++) { - if (removal_list_add(&state->removal_list, rem_list[i], current_time) < 0) { - sys_free(add_list); - sys_free(rem_list); - sys_free(missing); - return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); - } - } - - sys_free(add_list); - sys_free(rem_list); - - // Respond to the metadata server - if (num_missing == 0) { - - // No missing chunks, send success - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_SUCCESS); - if (!message_writer_free(&writer)) + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - } else { - - // Some chunks are missing, send error with missing list - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_ERROR); - - // Write error message - string error_msg = S("Missing chunks"); - uint16_t error_len = (uint16_t)error_msg.len; - message_write(&writer, &error_len, sizeof(error_len)); - message_write(&writer, error_msg.ptr, error_msg.len); - - // Write missing count and missing hashes - uint32_t tmp = num_missing; - message_write(&writer, &tmp, sizeof(tmp)); - message_write(&writer, missing, num_missing * sizeof(SHA256)); - - sys_free(missing); - - if (!message_writer_free(&writer)) - return -1; + if (timed_hash_set_insert(&state->cs_rem_list, hash, current_time) < 0) { + assert(0); // TODO + } } + if (binary_read(&reader, NULL, 1)) + return -1; + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_3); + + uint32_t count = tmp_list.count + state->cs_lst_list.count; // TODO: overflow + message_write(&writer, &count, sizeof(count)); + + for (int i = 0; i < tmp_list.count; i++) { + SHA256 hash = tmp_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + for (int i = 0; i < state->cs_lst_list.count; i++) { + SHA256 hash = state->cs_lst_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + if (!message_writer_free(&writer)) + return -1; + return 0; } static int -process_metadata_server_download_locations(ChunkServer *state, int conn_idx, ByteView msg) +process_metadata_server_sync_4(ChunkServer *state, int conn_idx, ByteView msg) { (void) conn_idx; @@ -566,14 +458,9 @@ static int process_metadata_server_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) { switch (type) { - - case MESSAGE_TYPE_STATE_UPDATE: - return process_metadata_server_state_update(state, conn_idx, msg); - - case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: - return process_metadata_server_download_locations(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_2: return process_metadata_server_sync_2(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_4: return process_metadata_server_sync_4(state, conn_idx, msg); } - return -1; } @@ -884,6 +771,32 @@ start_connecting_to_metadata_server(ChunkServer *state) return 0; } +static int send_sync_message(ChunkServer *state) +{ + assert(state->disconnect_time == INVALID_TIME); + + int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER); + assert(conn_idx > -1); + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC); + + uint32_t count = state->cs_add_list.count; // TODO: check implicit conversions + message_write(&writer, &count, sizeof(count)); + + for (uint32_t i = 0; i < count; i++) { + SHA256 hash = state->cs_add_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + if (!message_writer_free(&writer)) + return -1; + return 0; +} + int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout) { string addr = getargs(argc, argv, "--addr", "127.0.0.1"); @@ -899,6 +812,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts if (remote_port <= 0 || remote_port >= 1<<16) return -1; + Time current_time = get_current_time(); + state->trace = trace; tcp_context_init(&state->tcp); @@ -917,7 +832,9 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts state->downloading = false; pending_download_list_init(&state->pending_download_list); - removal_list_init(&state->removal_list); + hash_set_init(&state->cs_add_list); + hash_set_init(&state->cs_lst_list); + timed_hash_set_init(&state->cs_rem_list); char tmp[1<<10]; if (addr.len >= (int) sizeof(tmp)) { @@ -950,9 +867,12 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts } state->remote_addr.port = remote_port; state->disconnect_time = INVALID_TIME; + state->last_sync_time = current_time; start_connecting_to_metadata_server(state); + // TODO: add all chunk hashes to the add list + printf("Chunk server set up (local=%.*s:%d, remote=%.*s:%d, path=%.*s)\n", addr.len, addr.ptr, @@ -971,7 +891,9 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts int chunk_server_free(ChunkServer *state) { pending_download_list_free(&state->pending_download_list); - removal_list_free(&state->removal_list); + timed_hash_set_free(&state->cs_rem_list); + hash_set_free(&state->cs_lst_list); + hash_set_free(&state->cs_add_list); chunk_store_free(&state->store); tcp_context_free(&state->tcp); return 0; @@ -991,14 +913,16 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled switch (events[i].type) { case EVENT_CONNECT: - if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) - state->disconnect_time = INVALID_TIME; + if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) { + assert(state->disconnect_time == INVALID_TIME); + } break; case EVENT_DISCONNECT: - switch (tcp_get_tag(&state->tcp, conn_idx)) { + switch (events[i].tag) { case TAG_METADATA_SERVER: + assert(state->disconnect_time == INVALID_TIME); state->disconnect_time = current_time; break; @@ -1057,20 +981,37 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled } } + { + int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER); + assert((conn_idx < 0) == (state->disconnect_time != INVALID_TIME)); + } + Time deadline = INVALID_TIME; // Remove items from the remove list that got too old - for (int i = 0; i < state->removal_list.count; i++) { - PendingRemoval *removal = &state->removal_list.items[i]; - Time removal_time = removal->marked_time + (Time) DELETION_TIMEOUT * 1000000000; + for (int i = 0; i < state->cs_rem_list.count; i++) { + TimedHash *removal = &state->cs_rem_list.items[i]; + Time removal_time = removal->time + (Time) DELETION_TIMEOUT * 1000000000; if (removal_time < current_time) { if (chunk_store_remove(&state->store, removal->hash) == 0) - *removal = state->removal_list.items[--state->removal_list.count]; + *removal = state->cs_rem_list.items[--state->cs_rem_list.count]; } else { nearest_deadline(&deadline, removal_time); } } + if (state->disconnect_time == INVALID_TIME) { + Time next_sync_time = state->last_sync_time + (Time) SYNC_INTERVAL * 1000000000; + if (current_time >= next_sync_time) { + if (send_sync_message(state) < 0) { + assert(0); // TODO + } + state->last_sync_time = current_time; + } else { + nearest_deadline(&deadline, next_sync_time); + } + } + // TODO: periodically look for chunks that have their hashes messed up and delete them // Periodically retry pending downloads diff --git a/src/chunk_server.h b/src/chunk_server.h index 9e13331..34e89b8 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -3,6 +3,7 @@ #include +#include "metadata_server.h" #include "tcp.h" #define TAG_METADATA_SERVER 1 @@ -26,26 +27,33 @@ typedef struct { } PendingDownloadList; typedef struct { - SHA256 hash; - Time marked_time; -} PendingRemoval; -typedef struct { - int count; - int capacity; - PendingRemoval *items; -} RemovalList; + bool trace; + + Address local_addr; + + Address remote_addr; + + Time disconnect_time; + Time last_sync_time; + + TCP tcp; -typedef struct { - bool trace; - Address local_addr; - Address remote_addr; - Time disconnect_time; - TCP tcp; ChunkStore store; + bool downloading; + PendingDownloadList pending_download_list; - RemovalList removal_list; + + // List of chunks added since the last update + HashSet cs_add_list; + + // List of chunks marked for removal after a timeout + TimedHashSet cs_rem_list; + + // List of chunks that were lost due to errors or forceful removals of chunk files + HashSet cs_lst_list; + } ChunkServer; int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout); diff --git a/src/file_system.c b/src/file_system.c index c64f630..8f23a6c 100644 --- a/src/file_system.c +++ b/src/file_system.c @@ -268,3 +268,95 @@ int file_read_all(string path, string *data) file_close(fd); return 0; } + +#ifdef _WIN32 + +int directory_scanner_init(DirectoryScanner *scanner, string path) +{ + char pattern[PATH_MAX]; + int ret = snprintf(pattern, sizeof(pattern), "%.*s\\*", path.len, path.ptr); + if (ret < 0 || ret >= sizeof(pattern)) + return -1; + + scanner->handle = sys_FindFirstFileA(pattern, &scanner->find_data); + if (scanner->handle == INVALID_HANDLE_VALUE) { + if (GetLastError() == ERROR_FILE_NOT_FOUND) { + scanner->done = true; + return 0; + } + return -1; + } + + scanner->done = false; + scanner->first = true; + return 0; +} + +int directory_scanner_next(DirectoryScanner *scanner, string *name) +{ + if (scanner->done) + return 1; + + if (!scanner->first) { + BOOL ok = sys_FindNextFileA(scanner->handle, &scanner->find_data); + if (!ok) { + scanner->done = true; + if (GetLastError() == ERROR_NO_MORE_FILES) + return 1; + return -1; + } + } else { + scanner->first = false; + } + + char *p = scanner->find_data.cFileName; + *name = (string) { p, strlen(p) }; + return 0; +} + +void directory_scanner_free(DirectoryScanner *scanner) +{ + sys_FindClose(scanner->handle); +} + +#else + +int directory_scanner_init(DirectoryScanner *scanner, string path) +{ + char path_copy[PATH_MAX]; + if (path.len >= PATH_MAX) + return -1; + memcpy(path_copy, path.ptr, path.len); + path_copy[path.len] = '\0'; + + scanner->d = sys_opendir(path_copy); + if (scanner->d == NULL) { + scanner->done = true; + return -1; + } + + scanner->done = false; + return 0; +} + +int directory_scanner_next(DirectoryScanner *scanner, string *name) +{ + if (scanner->done) + return 1; + + scanner->e = sys_readdir(scanner->d); + if (scanner->e == NULL) { + scanner->done = true; + return 1; + } + + *name = (string) { scanner->e->d_name, strlen(scanner->e->d_name) }; + return 0; +} + +void directory_scanner_free(DirectoryScanner *scanner) +{ + sys_closedir(scanner->d); +} + +#endif diff --git a/src/file_system.h b/src/file_system.h index 125388d..a6ad0de 100644 --- a/src/file_system.h +++ b/src/file_system.h @@ -3,10 +3,29 @@ #include "basic.h" +#ifndef _WIN32 +#include +#endif + typedef struct { uint64_t data; } Handle; +#ifdef _WIN32 +typedef struct { + HANDLE handle; + WIN32_FIND_DATA find_data; + bool first; + bool done; +} DirectoryScanner; +#else +typedef struct { + DIR *d; + struct dirent *e; + bool done; +} DirectoryScanner; +#endif + int file_open(string path, Handle *fd); void file_close(Handle fd); int file_lock(Handle fd); @@ -22,4 +41,8 @@ int remove_file_or_dir(string path); int get_full_path(string path, char *dst); int file_read_all(string path, string *data); +int directory_scanner_init(DirectoryScanner *scanner, string path); +int directory_scanner_next(DirectoryScanner *scanner, string *name); +void directory_scanner_free(DirectoryScanner *scanner); + #endif // FILE_SYSTEM_INCLUDED diff --git a/src/hash_set.c b/src/hash_set.c new file mode 100644 index 0000000..a2ea582 --- /dev/null +++ b/src/hash_set.c @@ -0,0 +1,168 @@ +#include +#include + +#include "system.h" +#include "hash_set.h" + +void hash_set_init(HashSet *set) +{ + set->items = NULL; + set->count = 0; + set->capacity = 0; +} + +void hash_set_free(HashSet *set) +{ + sys_free(set->items); + set->items = NULL; +} + +void hash_set_clear(HashSet *set) +{ + free(set->items); + set->items = NULL; + set->count = 0; + set->capacity = 0; +} + +int hash_set_insert(HashSet *set, SHA256 hash) +{ + // Avoid duplicates + for (int i = 0; i < set->count; i++) + if (!memcmp(&set->items[i], &hash, sizeof(SHA256))) + return 0; // Already present + + if (set->count == set->capacity) { + + int new_capacity; + if (set->items == NULL) + new_capacity = 16; + else + new_capacity = 2 * set->capacity; + + SHA256 *new_items = sys_realloc(set->items, new_capacity * sizeof(SHA256)); + if (new_items == NULL) + return -1; + + set->items = new_items; + set->capacity = new_capacity; + } + + set->items[set->count++] = hash; + return 0; +} + +bool hash_set_remove(HashSet *set, SHA256 hash) +{ + for (int i = 0; i < set->count; i++) + if (!memcmp(&hash, &set->items[i], sizeof(SHA256))) { + set->items[i] = set->items[--set->count]; + return true; + } + return false; +} + +bool hash_set_contains(HashSet *set, SHA256 hash) +{ + for (int i = 0; i < set->count; i++) + if (!memcmp(&hash, &set->items[i], sizeof(SHA256))) + return true; + return false; +} + +int hash_set_merge(HashSet *dst, HashSet src) +{ + HashSet ret; + hash_set_init(&ret); + + for (int i = 0; i < dst->count; i++) { + if (hash_set_insert(&ret, dst->items[i]) < 0) + goto error; + } + + for (int i = 0; i < src.count; i++) { + if (hash_set_insert(&ret, src.items[i]) < 0) + goto error; + } + + hash_set_free(dst); + *dst = ret; + return 0; + +error: + hash_set_free(&ret); + return -1; +} + +void hash_set_remove_set(HashSet *dst, HashSet src) +{ + for (int i = 0; i < src.count; i++) + hash_set_remove(dst, src.items[i]); +} + +void timed_hash_set_init(TimedHashSet *set) +{ + set->items = NULL; + set->count = 0; + set->capacity = 0; +} + +void timed_hash_set_free(TimedHashSet *set) +{ + sys_free(set->items); + set->items = NULL; +} + +int timed_hash_set_find(TimedHashSet *set, SHA256 hash) +{ + for (int i = 0; i < set->count; i++) + if (!memcmp(&set->items[i].hash, &hash, sizeof(SHA256))) + return i; + return -1; +} + +int timed_hash_set_insert(TimedHashSet *set, SHA256 hash, Time time) +{ + // Check if already in set + int idx = timed_hash_set_find(set, hash); + if (idx >= 0) { + // Already marked, keep the original time + return 0; + } + + if (set->count == set->capacity) { + int new_capacity; + if (set->capacity == 0) + new_capacity = 8; + else + new_capacity = 2 * set->capacity; + + TimedHash *new_items = sys_malloc(new_capacity * sizeof(TimedHash)); + if (new_items == NULL) + return -1; + + if (set->capacity > 0) { + memcpy(new_items, set->items, set->count * sizeof(set->items[0])); + sys_free(set->items); + } + + set->items = new_items; + set->capacity = new_capacity; + } + + set->items[set->count++] = (TimedHash) { hash, time }; + return 0; +} + +void timed_hash_set_remove(TimedHashSet *set, SHA256 hash) +{ + int idx = timed_hash_set_find(set, hash); + if (idx >= 0) { + // Remove by shifting remaining items + if (idx < set->count - 1) { + memmove(&set->items[idx], &set->items[idx + 1], + (set->count - idx - 1) * sizeof(set->items[0])); + } + set->count--; + } +} diff --git a/src/hash_set.h b/src/hash_set.h new file mode 100644 index 0000000..31834d5 --- /dev/null +++ b/src/hash_set.h @@ -0,0 +1,38 @@ +#ifndef HASH_SET_INCLUDED +#define HASH_SET_INCLUDED + +#include "basic.h" + +typedef struct { + SHA256 *items; + int count; + int capacity; +} HashSet; + +typedef struct { + SHA256 hash; + Time time; +} TimedHash; + +typedef struct { + TimedHash *items; + int count; + int capacity; +} TimedHashSet; + +void hash_set_init (HashSet *set); +void hash_set_free (HashSet *set); +void hash_set_clear (HashSet *set); +int hash_set_insert (HashSet *set, SHA256 hash); +bool hash_set_remove (HashSet *set, SHA256 hash); +int hash_set_merge (HashSet *dst, HashSet src); +void hash_set_remove_set(HashSet *dst, HashSet src); +bool hash_set_contains (HashSet *set, SHA256 hash); + +void timed_hash_set_init (TimedHashSet *set); +void timed_hash_set_free (TimedHashSet *set); +int timed_hash_set_find (TimedHashSet *set, SHA256 hash); +int timed_hash_set_insert (TimedHashSet *set, SHA256 hash, Time time); +void timed_hash_set_remove (TimedHashSet *set, SHA256 hash); + +#endif // HASH_SET_INCLUDED diff --git a/src/main_test.c b/src/main_test.c index 294a0e7..17d0cfb 100644 --- a/src/main_test.c +++ b/src/main_test.c @@ -8,12 +8,20 @@ static sig_atomic_t simulation_should_stop = false; +static void signal_handler(int signum) +{ + (void)signum; + simulation_should_stop = true; +} + int main(int argc, char **argv) { (void)argc; (void)argv; - // TODO: set simulation_should_stop=true on ctrl+C + // Set up signal handlers for clean shutdown + signal(SIGINT, signal_handler); + signal(SIGTERM, signal_handler); startup_simulation(2); diff --git a/src/message.c b/src/message.c index bbe6738..cfdcb49 100644 --- a/src/message.c +++ b/src/message.c @@ -94,13 +94,14 @@ static char *message_type_to_str(uint16_t type) case MESSAGE_TYPE_WRITE_SUCCESS: return "WRITE_SUCCESS"; // Metadata server -> Chunk server - case MESSAGE_TYPE_STATE_UPDATE: return "STATE_UPDATE"; + case MESSAGE_TYPE_SYNC_2: return "SYNC_2"; + case MESSAGE_TYPE_SYNC_4: return "SYNC_4"; case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: return "DOWNLOAD_LOCATIONS"; // Chunk server -> Metadata server case MESSAGE_TYPE_AUTH: return "AUTH"; - case MESSAGE_TYPE_STATE_UPDATE_ERROR: return "UPDATE_ERROR"; - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: return "UPDATE_SUCCESS"; + case MESSAGE_TYPE_SYNC: return "SYNC"; + case MESSAGE_TYPE_SYNC_3: return "SYNC_3"; // Chunk server -> Client case MESSAGE_TYPE_CREATE_CHUNK_ERROR: return "CREATE_CHUNK_ERROR"; @@ -418,99 +419,7 @@ void message_dump(FILE *stream, ByteView msg) } break; - case MESSAGE_TYPE_DOWNLOAD_CHUNK: - printf(" (TODO)\n"); - break; - - // Metadata server -> Client - - case MESSAGE_TYPE_CREATE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_CREATE_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DELETE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DELETE_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_LIST_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_LIST_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_READ_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_READ_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_WRITE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_WRITE_SUCCESS: - printf(" (TODO)\n"); - break; - - // Metadata server -> Chunk server - - case MESSAGE_TYPE_STATE_UPDATE: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: - printf(" (TODO)\n"); - break; - - // Chunk server -> Metadata server - - case MESSAGE_TYPE_AUTH: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_STATE_UPDATE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: - printf(" (TODO)\n"); - break; - - // Chunk server -> Client - - case MESSAGE_TYPE_CREATE_CHUNK_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_CREATE_CHUNK_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_UPLOAD_CHUNK_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS: + default: printf(" (TODO)\n"); break; } diff --git a/src/message.h b/src/message.h index 5e943d3..641babb 100644 --- a/src/message.h +++ b/src/message.h @@ -33,13 +33,14 @@ enum { MESSAGE_TYPE_WRITE_SUCCESS, // Metadata server -> Chunk server - MESSAGE_TYPE_STATE_UPDATE, + MESSAGE_TYPE_SYNC_2, + MESSAGE_TYPE_SYNC_4, MESSAGE_TYPE_DOWNLOAD_LOCATIONS, // Chunk server -> Metadata server MESSAGE_TYPE_AUTH, - MESSAGE_TYPE_STATE_UPDATE_ERROR, - MESSAGE_TYPE_STATE_UPDATE_SUCCESS, + MESSAGE_TYPE_SYNC, + MESSAGE_TYPE_SYNC_3, // Chunk server -> Client MESSAGE_TYPE_CREATE_CHUNK_ERROR, diff --git a/src/metadata_server.c b/src/metadata_server.c index f230cf1..4338b7f 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -7,82 +7,36 @@ #include "message.h" #include "metadata_server.h" -static void hash_list_init(HashList *hash_list) -{ - hash_list->count = 0; - hash_list->capacity = 0; - hash_list->items = NULL; -} - -static void hash_list_free(HashList *hash_list) -{ - sys_free(hash_list->items); -} - -static int hash_list_insert(HashList *hash_list, SHA256 hash) -{ - // Avoid duplicates - for (int i = 0; i < hash_list->count; i++) - if (!memcmp(&hash_list->items[i], &hash, sizeof(SHA256))) - return 0; // Already present - - if (hash_list->count == hash_list->capacity) { - - int new_capacity; - if (hash_list->items == NULL) - new_capacity = 16; - else - new_capacity = 2 * hash_list->capacity; - - SHA256 *new_items = sys_realloc(hash_list->items, new_capacity * sizeof(SHA256)); - if (new_items == NULL) - return -1; - - hash_list->items = new_items; - hash_list->capacity = new_capacity; - } - - hash_list->items[hash_list->count++] = hash; - return 0; -} - -static bool hash_list_contains(HashList *hash_list, SHA256 hash) -{ - for (int j = 0; j < hash_list->count; j++) - if (!memcmp(&hash, &hash_list->items[j], sizeof(SHA256))) - return true; - return false; -} - static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_time) { chunk_server->used = true; chunk_server->auth = false; chunk_server->num_addrs = 0; - hash_list_init(&chunk_server->old_list); - hash_list_init(&chunk_server->add_list); - hash_list_init(&chunk_server->rem_list); + hash_set_init(&chunk_server->ms_old_list); + hash_set_init(&chunk_server->ms_add_list); + hash_set_init(&chunk_server->ms_rem_list); chunk_server->last_sync_time = current_time; chunk_server->last_response_time = current_time; } static void chunk_server_peer_free(ChunkServerPeer *chunk_server) { - hash_list_free(&chunk_server->rem_list); - hash_list_free(&chunk_server->add_list); - hash_list_free(&chunk_server->old_list); + hash_set_free(&chunk_server->ms_rem_list); + hash_set_free(&chunk_server->ms_add_list); + hash_set_free(&chunk_server->ms_old_list); chunk_server->used = false; } static bool chunk_server_peer_contains(ChunkServerPeer *chunk_server, SHA256 hash) { - return hash_list_contains(&chunk_server->old_list, hash) - || hash_list_contains(&chunk_server->add_list, hash); + return hash_set_contains(&chunk_server->ms_old_list, hash) + || hash_set_contains(&chunk_server->ms_add_list, hash); } static bool chunk_server_peer_load(ChunkServerPeer *chunk_server) { - return chunk_server->old_list.count + chunk_server->add_list.count; + return chunk_server->ms_old_list.count + + chunk_server->ms_add_list.count; } // Returns all chunk servers holding the given chunk @@ -152,7 +106,7 @@ static int find_chunk_server_by_addr(MetadataServer *state, Address addr) for (int i = 0; i < state->num_chunk_servers; i++) for (int j = 0; j < state->chunk_servers[i].num_addrs; j++) if (addr_eql(state->chunk_servers[i].addrs[j], addr)) - return j; + return i; return -1; } @@ -284,6 +238,7 @@ process_client_delete(MetadataServer *state, int conn_idx, ByteView msg) if (binary_read(&reader, NULL, 1)) return -1; + // TODO: return unused hashes and add them to the ms_rem_list of holder chunk servers int ret = file_tree_delete_entity(&state->file_tree, path); if (ret < 0) { @@ -671,7 +626,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) int k = find_chunk_server_by_addr(state, results[i].addrs[j]); if (k == -1) return -1; - if (hash_list_insert(&state->chunk_servers[k].add_list, new_hashes[i]) < 0) + if (hash_set_insert(&state->chunk_servers[k].ms_add_list, new_hashes[i]) < 0) return -1; } } @@ -684,7 +639,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) // Add to rem_list for all chunk servers that have this chunk for (int j = 0; j < state->num_chunk_servers; j++) { if (chunk_server_peer_contains(&state->chunk_servers[j], removed_hash)) { - if (!hash_list_insert(&state->chunk_servers[j].rem_list, removed_hash)) + if (!hash_set_insert(&state->chunk_servers[j].ms_rem_list, removed_hash)) return -1; } } @@ -726,36 +681,6 @@ chunk_server_from_conn(MetadataServer *state, int conn_idx) return &state->chunk_servers[tag]; } -static int send_state_update(MetadataServer *state, int chunk_server_idx) -{ - ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; - - int conn_idx = tcp_index_from_tag(&state->tcp, chunk_server_idx); - assert(conn_idx > -1); - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE); - - uint32_t add_count = chunk_server->add_list.count; - uint32_t rem_count = chunk_server->rem_list.count; - message_write(&writer, &add_count, sizeof(add_count)); - message_write(&writer, &rem_count, sizeof(rem_count)); - - for (uint32_t i = 0; i < add_count; i++) - message_write(&writer, &chunk_server->add_list.items[i], sizeof(SHA256)); - - for (uint32_t i = 0; i < rem_count; i++) - message_write(&writer, &chunk_server->rem_list.items[i], sizeof(SHA256)); - - if (!message_writer_free(&writer)) - return -1; - - return 0; -} - static int process_chunk_server_auth(MetadataServer *state, int conn_idx, ByteView msg) { @@ -768,55 +693,49 @@ static int process_chunk_server_auth(MetadataServer *state, if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - // Read IPv4s - { - uint32_t num_ipv4; - if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) + uint32_t num_ipv4; + if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) + return -1; + + for (uint32_t i = 0; i < num_ipv4; i++) { + + IPv4 ipv4; + if (!binary_read(&reader, &ipv4, sizeof(ipv4))) return -1; - for (uint32_t i = 0; i < num_ipv4; i++) { + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) + return -1; - IPv4 ipv4; - if (!binary_read(&reader, &ipv4, sizeof(ipv4))) - return -1; - - uint16_t port; - if (!binary_read(&reader, &port, sizeof(port))) - return -1; - - if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { - Address addr = {0}; - addr.ipv4 = ipv4; - addr.is_ipv4 = true; - addr.port = port; - chunk_server->addrs[chunk_server->num_addrs++] = addr; - } + if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { + Address addr = {0}; + addr.ipv4 = ipv4; + addr.is_ipv4 = true; + addr.port = port; + chunk_server->addrs[chunk_server->num_addrs++] = addr; } } - // Read IPv6s - { - uint32_t num_ipv6; - if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) + uint32_t num_ipv6; + if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) + return -1; + + for (uint32_t i = 0; i < num_ipv6; i++) { + + IPv6 ipv6; + if (!binary_read(&reader, &ipv6, sizeof(ipv6))) return -1; - for (uint32_t i = 0; i < num_ipv6; i++) { + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) + return -1; - IPv6 ipv6; - if (!binary_read(&reader, &ipv6, sizeof(ipv6))) - return -1; - - uint16_t port; - if (!binary_read(&reader, &port, sizeof(port))) - return -1; - - if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { - Address addr = {0}; - addr.ipv6 = ipv6; - addr.is_ipv4 = false; - addr.port = port; - chunk_server->addrs[chunk_server->num_addrs++] = addr; - } + if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { + Address addr = {0}; + addr.ipv6 = ipv6; + addr.is_ipv4 = false; + addr.port = port; + chunk_server->addrs[chunk_server->num_addrs++] = addr; } } @@ -833,90 +752,135 @@ static int process_chunk_server_auth(MetadataServer *state, // we accept all connections that provide valid address information. chunk_server->auth = true; + // No need to respond return 0; } -static int process_chunk_server_state_update_success(MetadataServer *state, +static int process_chunk_server_sync(MetadataServer *state, int conn_idx, ByteView msg) { - (void) msg; // Success message has no body - ChunkServerPeer *chunk_server = chunk_server_from_conn(state, conn_idx); + int chunk_server_idx = tcp_get_tag(&state->tcp, conn_idx); + assert(chunk_server_idx > -1); + assert(chunk_server_idx <= MAX_CHUNK_SERVERS); - // Merge add_list into old_list - for (int i = 0; i < chunk_server->add_list.count; i++) { - if (!hash_list_contains(&chunk_server->old_list, chunk_server->add_list.items[i])) - hash_list_insert(&chunk_server->old_list, chunk_server->add_list.items[i]); - } + ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; + assert(chunk_server->used); - // Clear add_list and rem_list - chunk_server->add_list.count = 0; - chunk_server->rem_list.count = 0; - - if (state->trace) { - int tag = tcp_get_tag(&state->tcp, conn_idx); - printf("Received STATE_UPDATE_SUCCESS from chunk server %d\n", tag); - } - - return 0; -} - -static int process_chunk_server_state_update_error(MetadataServer *state, - int conn_idx, ByteView msg) -{ BinaryReader reader = { msg.ptr, msg.len, 0 }; - // Read header if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - // Read error message - uint16_t error_len; - if (!binary_read(&reader, &error_len, sizeof(error_len))) + uint32_t count; + if (!binary_read(&reader, &count, sizeof(count))) return -1; - char error_msg[256]; - int read_len = error_len < sizeof(error_msg) - 1 ? error_len : sizeof(error_msg) - 1; - if (!binary_read(&reader, error_msg, read_len)) - return -1; - error_msg[read_len] = '\0'; + for (uint32_t i = 0; i < count; i++) { - // Skip remaining error message if it was too long - if (error_len > read_len) - binary_read(&reader, NULL, error_len - read_len); - - // Read missing chunks - uint32_t num_missing; - if (!binary_read(&reader, &num_missing, sizeof(num_missing))) - return -1; - - SHA256 *missing_chunks = sys_malloc(num_missing * sizeof(SHA256)); - if (missing_chunks == NULL) - return -1; - - for (uint32_t i = 0; i < num_missing; i++) { - if (!binary_read(&reader, &missing_chunks[i], sizeof(SHA256))) { - sys_free(missing_chunks); + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) return -1; + + // If the chunk is not referenced by the file tree, do + // nothing. + + if (!file_tree_uses_hash(&state->file_tree, hash)) + continue; + + // If the chunk is properly replicated or under-replicated, + // add it to the ms_add_list. + + int holders[MAX_CHUNK_SERVERS]; + int num_holders = all_chunk_servers_holding_chunk(state, hash, holders, MAX_CHUNK_SERVERS); + assert(num_holders > -1); + assert(num_holders <= MAX_CHUNK_SERVERS); + + if (num_holders <= state->replication_factor) { + if (hash_set_insert(&chunk_server->ms_add_list, hash) < 0) { + assert(0); // TODO + } + continue; } + + // If the chunk is over-replicated, either don't add + // it to the ms_add_list or add it to the ms_rem_list + // of some other holder. + // + // TODO: For now we don't add it to the ms_add_list, + // but there may be a better solution. } - if (state->trace) { - int tag = tcp_get_tag(&state->tcp, conn_idx); - printf("Received STATE_UPDATE_ERROR from chunk server %d: %s (missing %u chunks)\n", - tag, error_msg, num_missing); - } + if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion + return -1; + + // Respond with ms_add_list and ms_rem_list ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); assert(output); MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_LOCATIONS); + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_2); - message_write(&writer, &num_missing, sizeof(num_missing)); + uint32_t add_count = chunk_server->ms_add_list.count; // TODO: check implicit casts + message_write(&writer, &add_count, sizeof(add_count)); - for (uint32_t i = 0; i < num_missing; i++) { + for (uint32_t i = 0; i < add_count; i++) { + SHA256 hash = chunk_server->ms_add_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } - SHA256 hash = missing_chunks[i]; + uint32_t rem_count = chunk_server->ms_rem_list.count; // TODO: check implicit casts + message_write(&writer, &rem_count, sizeof(rem_count)); + + for (uint32_t i = 0; i < rem_count; i++) { + SHA256 hash = chunk_server->ms_rem_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + if (!message_writer_free(&writer)) + return -1; + return 0; +} + +static int process_chunk_server_sync_3(MetadataServer *state, + int conn_idx, ByteView msg) +{ + int chunk_server_idx = tcp_get_tag(&state->tcp, conn_idx); + assert(chunk_server_idx > -1); + assert(chunk_server_idx <= MAX_CHUNK_SERVERS); + + ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; + assert(chunk_server->used); + + BinaryReader reader = { msg.ptr, msg.len, 0 }; + + if (!binary_read(&reader, NULL, sizeof(MessageHeader))) + return -1; + + uint32_t count; + if (!binary_read(&reader, &count, sizeof(count))) + return -1; + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_4); + + HashSet tmp_list; + hash_set_init(&tmp_list); + + message_write(&writer, &count, sizeof(count)); + + for (uint32_t i = 0; i < count; i++) { + + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) + return -1; + + if (hash_set_insert(&tmp_list, hash) < 0) { + assert(0); // TODO + } int holders[MAX_CHUNK_SERVERS]; int num_holders = all_chunk_servers_holding_chunk(state, hash, holders, MAX_CHUNK_SERVERS); @@ -926,17 +890,26 @@ static int process_chunk_server_state_update_error(MetadataServer *state, uint32_t tmp = num_holders; message_write(&writer, &tmp, sizeof(tmp)); - for (int j = 0; j < num_holders; j++) - message_write_server_addr(&writer, &state->chunk_servers[holders[j]]); - - message_write(&writer, &hash, sizeof(hash)); + for (int j = 0; j < num_holders; j++) { + int k = holders[j]; + message_write_server_addr(&writer, &state->chunk_servers[k]); + } } - sys_free(missing_chunks); + if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion + return -1; + + if (hash_set_merge(&chunk_server->ms_old_list, chunk_server->ms_add_list) < 0) { + assert(0); // TODO + } + + hash_set_remove_set(&chunk_server->ms_old_list, tmp_list); + + hash_set_free(&chunk_server->ms_add_list); + chunk_server->ms_add_list = tmp_list; if (!message_writer_free(&writer)) return -1; - return 0; } @@ -948,11 +921,11 @@ process_chunk_server_message(MetadataServer *state, case MESSAGE_TYPE_AUTH: return process_chunk_server_auth(state, conn_idx, msg); - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: - return process_chunk_server_state_update_success(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC: + return process_chunk_server_sync(state, conn_idx, msg); - case MESSAGE_TYPE_STATE_UPDATE_ERROR: - return process_chunk_server_state_update_error(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_3: + return process_chunk_server_sync_3(state, conn_idx, msg); } return -1; } @@ -961,8 +934,8 @@ static bool is_chunk_server_message_type(uint16_t type) { switch (type) { case MESSAGE_TYPE_AUTH: - case MESSAGE_TYPE_STATE_UPDATE_ERROR: - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: + case MESSAGE_TYPE_SYNC: + case MESSAGE_TYPE_SYNC_3: return true; default: @@ -1025,7 +998,6 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * Event events[MAX_CONNS+1]; int num_events = tcp_translate_events(&state->tcp, events, contexts, polled, num_polled); - // Implement periodic health checks: send STATE_UPDATE to chunk servers Time current_time = get_current_time(); if (current_time == INVALID_TIME) { assert(0); // TODO @@ -1041,9 +1013,9 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * case EVENT_DISCONNECT: { - int tag = tcp_get_tag(&state->tcp, conn_idx); - if (tag >= 0) { - chunk_server_peer_free(&state->chunk_servers[tag]); + if (events[i].tag >= 0) { + chunk_server_peer_free(&state->chunk_servers[events[i].tag]); + assert(state->num_chunk_servers > 0); state->num_chunk_servers--; } } @@ -1079,6 +1051,7 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * } chunk_server_peer_init(&state->chunk_servers[j], current_time); + state->num_chunk_servers++; tcp_set_tag(&state->tcp, conn_idx, j, true); @@ -1119,23 +1092,10 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * Time response_timeout = chunk_server->last_response_time + (Time) RESPONSE_TIME_LIMIT * 1000000000; if (current_time > response_timeout) { - // TODO: drop the chunk server + assert(0); // TODO: drop the chunk server continue; } nearest_deadline(&next_wakeup, response_timeout); - - if (chunk_server->auth && chunk_server->last_sync_done) { - Time sync_timeout = chunk_server->last_sync_time + (Time) SYNC_INTERVAL * 1000000000; - if (current_time > sync_timeout) { - if (send_state_update(state, i) < 0) { - assert(0); // TODO - } - chunk_server->last_sync_time = current_time; - chunk_server->last_sync_done = false; - continue; - } - nearest_deadline(&next_wakeup, sync_timeout); - } } *timeout = deadline_to_timeout(next_wakeup, current_time); diff --git a/src/metadata_server.h b/src/metadata_server.h index 7edc499..d8b7ed8 100644 --- a/src/metadata_server.h +++ b/src/metadata_server.h @@ -5,16 +5,11 @@ #include "file_tree.h" #include "config.h" #include "basic.h" +#include "hash_set.h" #define CONNECTION_TAG_CLIENT -2 #define CONNECTION_TAG_UNKNOWN -3 -typedef struct { - int count; - int capacity; - SHA256 *items; -} HashList; - typedef struct { bool used; @@ -23,17 +18,14 @@ typedef struct { int num_addrs; Address addrs[MAX_SERVER_ADDRS]; - // Chunks held by the chunk server during - // the last update - HashList old_list; + // List of chunks that are known to be held by CS + HashSet ms_old_list; // TODO: rename all *_list symbols to *_set - // Chunks added to the chunk server since - // the last update - HashList add_list; + // List of chunks that should be held by CS + HashSet ms_add_list; - // Chunks removed from the chunk server - // since the last update - HashList rem_list; + // List of chunks that may be held by CS but should removed from it + HashSet ms_rem_list; // Time when last STATE_UPDATE was sent Time last_sync_time; diff --git a/src/system.c b/src/system.c index a98a999..9ae26c2 100644 --- a/src/system.c +++ b/src/system.c @@ -31,6 +31,7 @@ typedef enum { DESC_SOCKET, DESC_LISTEN_SOCKET, DESC_CONNECTION_SOCKET, + DESC_DIRECTORY, } DescriptorType; typedef enum { @@ -84,6 +85,14 @@ typedef struct { NATIVE_HANDLE real_fd; + // ------ Directory ------------- + +#ifdef _WIN32 + HANDLE real_d; +#else + DIR *real_d; +#endif + // ------ Socket ---------------- // Events reported by the last "poll" call @@ -562,152 +571,79 @@ static bool data_queue_full(DataQueue *queue) return queue->used == queue->size; } -static int compare_processes(const void *p1, const void *p2) +static int setup_poll_array(void **contexts, struct pollfd *polled) { - Process *a = *(Process**) p1; - Process *b = *(Process**) p2; - if (b->wakeup_time == INVALID_TIME) return -1; - if (a->wakeup_time == INVALID_TIME) return +1; - return a->wakeup_time - b->wakeup_time; -} + int num_polled = 0; -void update_simulation(void) -{ - // Order processes by wakeup time. Those with no - // wakeup time go last. - Process *ordered_processes[MAX_PROCESSES]; - for (int i = 0; i < num_processes; i++) - ordered_processes[i] = processes[i]; + for (int j = 0, k = 0; k < current_process->num_desc; j++) { - qsort(ordered_processes, num_processes, sizeof(*ordered_processes), compare_processes); + assert(j < MAX_DESCRIPTORS); + Descriptor *desc = ¤t_process->desc[j]; + if (desc->type == DESC_EMPTY) + continue; + k++; - for (int i = 0; i < num_processes; i++) { + int revents = 0; + switch (desc->type) { - current_process = ordered_processes[i]; + case DESC_FILE: + assert(0); // TODO: error + break; - void *contexts[MAX_CONNS+1]; - struct pollfd polled[MAX_CONNS+1]; - int num_polled = 0; + case DESC_SOCKET: + // Ignore + break; - for (int j = 0, k = 0; k < current_process->num_desc; j++) { + case DESC_LISTEN_SOCKET: + if (!accept_queue_empty(&desc->accept_queue)) + revents |= POLLIN; + break; - assert(j < MAX_DESCRIPTORS); - Descriptor *desc = ¤t_process->desc[j]; - if (desc->type == DESC_EMPTY) - continue; - k++; + case DESC_CONNECTION_SOCKET: + switch (desc->connection_state) { - int revents = 0; - switch (desc->type) { - - case DESC_FILE: - assert(0); // TODO: error + case CONNECTION_DELAYED: break; - case DESC_SOCKET: - // Ignore + case CONNECTION_QUEUED: break; - case DESC_LISTEN_SOCKET: - if (!accept_queue_empty(&desc->accept_queue)) - revents |= POLLIN; - break; - - case DESC_CONNECTION_SOCKET: - switch (desc->connection_state) { - - case CONNECTION_DELAYED: - break; - - case CONNECTION_QUEUED: - break; - - case CONNECTION_ESTABLISHED: - { - Descriptor *peer = handle_to_desc(desc->connection_peer); - if (peer == NULL) { + case CONNECTION_ESTABLISHED: + { + Descriptor *peer = handle_to_desc(desc->connection_peer); + if (peer == NULL) { + revents |= POLLIN; + } else { + if (!data_queue_full(&desc->output_data)) + revents |= POLLOUT; + if (!data_queue_empty(&peer->output_data)) revents |= POLLIN; - } else { - if (!data_queue_full(&desc->output_data)) - revents |= POLLOUT; - if (!data_queue_empty(&peer->output_data)) - revents |= POLLIN; - } } - break; - - case CONNECTION_FAILED: - assert(0); // TODO - break; } break; + + case CONNECTION_FAILED: + assert(0); // TODO + break; } - - revents &= desc->events; - if (revents) { - polled[num_polled].fd = (SOCKET) j; - polled[num_polled].events = desc->events; - polled[num_polled].revents = revents; - contexts[num_polled] = desc->context; - num_polled++; - } + break; } - if (num_polled == 0) { - - Time wakeup_time = current_process->wakeup_time; - if (wakeup_time == INVALID_TIME) continue; - - assert(current_time <= wakeup_time); - current_time = wakeup_time; + revents &= desc->events; + if (revents) { + polled[num_polled].fd = (SOCKET) j; + polled[num_polled].events = desc->events; + polled[num_polled].revents = revents; + contexts[num_polled] = desc->context; + num_polled++; } - - int timeout = -1; - switch (current_process->type) { - case PROCESS_TYPE_METADATA_SERVER: - num_polled = metadata_server_step( - ¤t_process->metadata_server, - contexts, - polled, - num_polled, - &timeout - ); - break; - case PROCESS_TYPE_CHUNK_SERVER: - num_polled = chunk_server_step( - ¤t_process->chunk_server, - contexts, - polled, - num_polled, - &timeout - ); - break; - case PROCESS_TYPE_CLIENT: - num_polled = simulation_client_step( - ¤t_process->simulation_client, - contexts, - polled, - num_polled, - &timeout - ); - break; - } - - if (num_polled < 0) { - // TODO - } - - if (timeout < 0) { - current_process->wakeup_time = INVALID_TIME; - } else { - current_process->wakeup_time = current_time + timeout * 1000000; - } - - process_poll_array(current_process, contexts, polled, num_polled); - - current_process = NULL; } + return num_polled; +} + +static void update_network(void) +{ for (int i = 0; i < num_processes; i++) { for (int j = 0, k = 0; k < processes[i]->num_desc; j++) { @@ -767,6 +703,142 @@ void update_simulation(void) } } +void update_simulation(void) +{ + for (;;) { + int num_ready = 0; + for (int i = 0; i < num_processes; i++) { + + current_process = processes[i]; + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled = setup_poll_array(contexts, polled); + + if (num_polled > 0) { + + num_ready++; + + int timeout = -1; + switch (current_process->type) { + case PROCESS_TYPE_METADATA_SERVER: + num_polled = metadata_server_step( + ¤t_process->metadata_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CHUNK_SERVER: + num_polled = chunk_server_step( + ¤t_process->chunk_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CLIENT: + num_polled = simulation_client_step( + ¤t_process->simulation_client, + contexts, + polled, + num_polled, + &timeout + ); + break; + } + + if (num_polled < 0) { + assert(0); // TODO + } + + if (timeout < 0) { + current_process->wakeup_time = INVALID_TIME; + } else { + current_process->wakeup_time = current_time + (Time) timeout * 1000000; + } + + process_poll_array(current_process, contexts, polled, num_polled); + } + + current_process = NULL; + + update_network(); + } + + if (num_ready == 0) + break; + } + + Process *next_process = NULL; + for (int i = 0; i < num_processes; i++) + if (processes[i]->wakeup_time != INVALID_TIME) + if (next_process == NULL || processes[i]->wakeup_time < next_process->wakeup_time) + next_process = processes[i]; + + if (next_process == NULL) { + assert(0); // Nothing to schedule next + } + + assert(current_time <= next_process->wakeup_time); + current_time = next_process->wakeup_time; + + current_process = next_process; + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled = setup_poll_array(contexts, polled); + + int timeout = -1; + switch (current_process->type) { + case PROCESS_TYPE_METADATA_SERVER: + num_polled = metadata_server_step( + ¤t_process->metadata_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CHUNK_SERVER: + num_polled = chunk_server_step( + ¤t_process->chunk_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CLIENT: + num_polled = simulation_client_step( + ¤t_process->simulation_client, + contexts, + polled, + num_polled, + &timeout + ); + break; + } + + if (num_polled < 0) { + assert(0); // TODO + } + + if (timeout < 0) { + current_process->wakeup_time = INVALID_TIME; + } else { + current_process->wakeup_time = current_time + (Time) timeout * 1000000; + } + + process_poll_array(current_process, contexts, polled, num_polled); + + current_process = NULL; + + update_network(); +} + void *mock_malloc(size_t len) { return malloc(len); @@ -1236,6 +1308,14 @@ static void close_desc(Descriptor *desc) } // TODO break; + + case DESC_DIRECTORY: +#ifdef _WIN32 + FindClose(desc->real_d); +#else + closedir(desc->real_d); +#endif + break; } desc->type = DESC_EMPTY; desc->generation++; @@ -1503,6 +1583,71 @@ int mock_ioctlsocket(SOCKET fd, long cmd, u_long *argp) return -1; } +HANDLE mock_FindFirstFileA(char *lpFileName, WIN32_FIND_DATAA *lpFindFileData) +{ + HANDLE handle = FindFirstFileA(lpFileName, lpFindFileData); + if (handle == INVALID_HANDLE_VALUE) + return INVALID_HANDLE_VALUE; + + if (current_process->num_desc == MAX_DESCRIPTORS) { + FindClose(handle); + SetLastError(ERROR_TOO_MANY_OPEN_FILES); + return INVALID_HANDLE_VALUE; + } + + int idx = 0; + while (current_process->desc[idx].type != DESC_EMPTY) { + idx++; + assert(idx < MAX_DESCRIPTORS); + } + + Descriptor *desc = ¤t_process->desc[idx]; + desc->type = DESC_DIRECTORY; + desc->real_d = handle; + + current_process->num_desc++; + CHECK_NON_EMPTY_DESC_INVARIANT; + return (HANDLE) idx; +} + +BOOL mock_FindNextFileA(HANDLE hFindFile, WIN32_FIND_DATAA *lpFindFileData) +{ + if (hFindFile == INVALID_HANDLE_VALUE || (int)hFindFile < 0 || (int)hFindFile >= MAX_DESCRIPTORS) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + int idx = (int) hFindFile; + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + + // Forward to real FindNextFileA, last error is set by the real call + return FindNextFileA(desc->real_d, lpFindFileData); +} + +BOOL mock_FindClose(HANDLE hFindFile) +{ + if (hFindFile == INVALID_HANDLE_VALUE || (int)hFindFile < 0 || (int)hFindFile >= MAX_DESCRIPTORS) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + int idx = (int) hFindFile; + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + + close_desc(desc); + current_process->num_desc--; + CHECK_NON_EMPTY_DESC_INVARIANT; + return TRUE; +} + #else int mock_clock_gettime(clockid_t clockid, struct timespec *tp) @@ -1736,6 +1881,81 @@ int mock_fcntl(int fd, int cmd, ...) return -1; } +DIR *mock_opendir(char *name) +{ + DIR *dir = opendir(name); + if (dir == NULL) + return NULL; + + if (current_process->num_desc == MAX_DESCRIPTORS) { + closedir(dir); + errno = EMFILE; // Too many open files + return NULL; + } + + int idx = 0; + while (current_process->desc[idx].type != DESC_EMPTY) { + idx++; + assert(idx < MAX_DESCRIPTORS); + } + + Descriptor *desc = ¤t_process->desc[idx]; + desc->type = DESC_DIRECTORY; + desc->real_d = dir; + + current_process->num_desc++; + CHECK_NON_EMPTY_DESC_INVARIANT; + return (DIR *)(intptr_t) idx; +} + +struct dirent *mock_readdir(DIR *dirp) +{ + if (dirp == NULL) { + errno = EBADF; // Bad file descriptor + return NULL; + } + int idx = (int)(intptr_t) dirp; + + if (idx < 0 || idx >= MAX_DESCRIPTORS) { + errno = EBADF; + return NULL; + } + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + errno = EBADF; + return NULL; + } + + // Forward to real readdir, errno is set by the real call + return readdir(desc->real_d); +} + +int mock_closedir(DIR *dirp) +{ + if (dirp == NULL) { + errno = EBADF; // Bad file descriptor + return -1; + } + int idx = (int)(intptr_t) dirp; + + if (idx < 0 || idx >= MAX_DESCRIPTORS) { + errno = EBADF; + return -1; + } + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + errno = EBADF; + return -1; + } + + close_desc(desc); + current_process->num_desc--; + CHECK_NON_EMPTY_DESC_INVARIANT; + return 0; +} + #endif // !_WIN32 #endif // BUILD_TEST diff --git a/src/system.h b/src/system.h index aeea168..31ba995 100644 --- a/src/system.h +++ b/src/system.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -69,6 +70,9 @@ BOOL mock_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount); BOOL mock_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency); char* mock__fullpath(char *path, char *dst, int cap); int mock__mkdir(char *path); +HANDLE mock_FindFirstFileA(char *lpFileName, WIN32_FIND_DATAA *lpFindFileData); +BOOL mock_FindNextFileA(HANDLE hFindFile, WIN32_FIND_DATAA *lpFindFileData); +BOOL mock_FindClose(HANDLE hFindFile); #else int mock_clock_gettime(clockid_t clockid, struct timespec *tp); int mock_open(char *path, int flags, int mode); @@ -82,6 +86,9 @@ int mock_mkstemp(char *path); char* mock_realpath(char *path, char *dst); int mock_mkdir(char *path, mode_t mode); int mock_fcntl(int fd, int cmd, ...); +DIR* mock_opendir(char *name); +struct dirent* mock_readdir(DIR *dirp); +int mock_closedir(DIR *dirp); #endif // Common @@ -114,6 +121,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys__fullpath mock__fullpath #define sys_QueryPerformanceCounter mock_QueryPerformanceCounter #define sys_QueryPerformanceFrequency mock_QueryPerformanceFrequency +#define sys_FindFirstFileA mock_FindFirstFileA +#define sys_FindNextFileA mock_FindNextFileA +#define sys_FindClose mock_FindClose // Linux #define sys_mkdir mock_mkdir @@ -129,6 +139,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys_clock_gettime mock_clock_gettime #define sys_fcntl mock_fcntl #define sys_getrandom mock_getrandom +#define sys_opendir mock_opendir +#define sys_readdir mock_readdir +#define sys_closedir mock_closedir #else @@ -162,6 +175,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys__fullpath _fullpath #define sys_QueryPerformanceCounter QueryPerformanceCounter #define sys_QueryPerformanceFrequency QueryPerformanceFrequency +#define sys_FindFirstFileA FindFirstFileA +#define sys_FindNextFileA FindNextFileA +#define sys_FindClose FindClose // Linux #define sys_mkdir mkdir @@ -176,6 +192,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys_realpath realpath #define sys_clock_gettime clock_gettime #define sys_fcntl fcntl +#define sys_opendir opendir +#define sys_readdir readdir +#define sys_closedir closedir #endif #endif // SYSTEM_INCLUDED diff --git a/src/tcp.c b/src/tcp.c index 1affff1..36f16eb 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -240,8 +240,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd if (set_socket_blocking(new_fd, false) < 0) CLOSE_SOCKET(new_fd); else { - events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns }; conn_init(&tcp->conns[tcp->num_conns++], new_fd, false); + events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns-1, tcp->conns[tcp->num_conns-1].tag }; } } } @@ -266,7 +266,7 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd defer_close = true; else { conn->connecting = false; - events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns }; + events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns, conn->tag }; } } @@ -319,8 +319,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd removed[i] = defer_close; if (0) {} - else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns }; - else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns }; + else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns, conn->tag }; + else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns, conn->tag }; } } diff --git a/src/tcp.h b/src/tcp.h index 015bc7d..2a41a19 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -23,6 +23,7 @@ typedef enum { typedef struct { EventType type; int conn_idx; + int tag; } Event; typedef struct {