cozis/ToastyFS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #11 from cozis/improve-chunk-management-policy

Browse Source 
Improve chunk management policy
This commit is contained in:
Francesco Cozzuto
2025-11-14 01:17:57 +01:00
committed by GitHub
parent d598ccedb2 9a944feb01
commit 787807031b
25 changed files with 1376 additions and 714 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitattributes
+2
View File
@@ -0,0 +1,2 @@
# Ensure shell scripts always use LF line endings
*.sh text eol=lf
.gitignore
+5
View File
@@ -3,3 +3,8 @@
*.o
*.a
chunk_server_data_*
# Coverage reports
coverage_report/
*.gcov
*.gcda
*.gcno
DESIGN.txt
+66 -28
View File
@@ -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.
Makefile
+30 -11
View File
@@ -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
TODO.txt
+8
View File
@@ -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
scripts/README.md
+48
View File
@@ -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`
scripts/fix_line_endings.sh
Executable
+18
View File
@@ -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"
scripts/generate_coverage_html.sh
Executable
+31
View File
@@ -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"
scripts/measure_coverage.sh
Executable
+113
View File
@@ -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}"
src/basic.c
+3 -3
View File
@@ -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;
}
src/chunk_server.c
+140 -199
View File
@@ -4,10 +4,12 @@
#include <assert.h>
#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
src/chunk_server.h
+23 -15
View File
@@ -3,6 +3,7 @@
#include <limits.h>
#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);
src/file_system.c
+92
View File
@@ -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
src/file_system.h
+23
View File
@@ -3,10 +3,29 @@
#include "basic.h"
#ifndef _WIN32
#include <dirent.h>
#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
src/hash_set.c
+168
View File
@@ -0,0 +1,168 @@
#include <assert.h>
#include <string.h>
#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--;
}
}
src/hash_set.h
+38
View File
@@ -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
src/main_test.c
+9 -1
View File
@@ -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);
src/message.c
+5 -96
View File
@@ -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;
}
src/message.h
+4 -3
View File
@@ -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,
src/metadata_server.c
+176 -216
View File
@@ -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);
src/metadata_server.h
+7 -15
View File
@@ -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;
src/system.c
+343 -123
View File
@@ -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 = &current_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 = &current_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(
&current_process->metadata_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CHUNK_SERVER:
num_polled = chunk_server_step(
&current_process->chunk_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CLIENT:
num_polled = simulation_client_step(
&current_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(
&current_process->metadata_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CHUNK_SERVER:
num_polled = chunk_server_step(
&current_process->chunk_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CLIENT:
num_polled = simulation_client_step(
&current_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(
&current_process->metadata_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CHUNK_SERVER:
num_polled = chunk_server_step(
&current_process->chunk_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CLIENT:
num_polled = simulation_client_step(
&current_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 = &current_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 = &current_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 = &current_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 = &current_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 = &current_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 = &current_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
src/system.h
+19
View File
@@ -21,6 +21,7 @@
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <dirent.h>
#include <sys/stat.h>
#include <sys/file.h>
#include <sys/socket.h>
@@ -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
src/tcp.c
+4 -4
View File
@@ -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 };
}
}
src/tcp.h
+1
View File
@@ -23,6 +23,7 @@ typedef enum {
typedef struct {
EventType type;
int conn_idx;
int tag;
} Event;
typedef struct {