Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 56 additions & 77 deletions .github/workflows/ci-test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -13,26 +13,29 @@ jobs:
unit-tests:
name: "Build ${{ matrix.build_type }} with ${{ matrix.compiler }}"
runs-on: ubuntu-latest
env:
CALF_LOG_DIR: ${{ github.workspace }}/capiocl_logs
strategy:
fail-fast: false
matrix:
build_type:
- Debug
- Release
compiler:
- gcc:9
- gcc:10
- gcc:11
- gcc:12
- gcc:13
- silkeh/clang:13
- silkeh/clang:14
- silkeh/clang:15
- silkeh/clang:16
- silkeh/clang:17
- silkeh/clang:18
- gcc:14
- gcc:15
- gcc:16
- zhongruoyu/llvm-ports:13-trixie
- zhongruoyu/llvm-ports:14-trixie
- zhongruoyu/llvm-ports:15-trixie
- zhongruoyu/llvm-ports:16-trixie
- zhongruoyu/llvm-ports:17-trixie
- zhongruoyu/llvm-ports:18-trixie
- zhongruoyu/llvm-ports:19-trixie
- zhongruoyu/llvm-ports:20-trixie
- zhongruoyu/llvm-ports:21-trixie
- zhongruoyu/llvm-ports:22-trixie
- zhongruoyu/llvm-ports:23-trixie

container:
image: ${{ matrix.compiler }}
Expand All @@ -48,75 +51,57 @@ jobs:
python3 python3-pip python3-venv \
libcurl4-gnutls-dev libprotobuf-dev protobuf-compiler

- name: "Set log directory"
shell: bash
run: echo "CALF_LOG_DIR=$GITHUB_WORKSPACE/capiocl_logs" >> "$GITHUB_ENV"

- name: "Run CMake"
shell: bash
run: |
cmake -DCMAKE_BUILD_TYPE=${{ matrix.build_type }} \
-DENABLE_COVERAGE=OFF \
-DENABLE_COVERAGE=${{ matrix.build_type == 'Debug' && matrix.compiler == 'gcc:16' && 'ON' || 'OFF' }} \
-DBUILD_PYTHON_BINDINGS=OFF \
-DCAPIO_CL_BUILD_TESTS=ON \
-G Ninja \
-B ${{ github.workspace }}/build \
.
cmake --build ${{ github.workspace }}/build -j $(nproc)
sudo cmake --install ${{ github.workspace }}/build --prefix /usr/local
-B "$GITHUB_WORKSPACE/build" \
-S "$GITHUB_WORKSPACE"
cmake --build "$GITHUB_WORKSPACE/build" -j $(nproc)
sudo cmake --install "$GITHUB_WORKSPACE/build" --prefix /usr/local

- name: Run unit tests
shell: bash
run: CAPIO_CL_tests --gtest_break_on_failure --gtest_print_time=1

- name: Zip logs
if: always()
- name: Generate coverage report
if: matrix.build_type == 'Debug' && matrix.compiler == 'gcc:16'
shell: bash
run: |
if [ -d "capiocl_logs" ]; then
cmake -E tar cf capiocl_logs.zip --format=zip capiocl_logs
fi
python3 -m venv /tmp/gcovr
/tmp/gcovr/bin/pip install gcovr
/tmp/gcovr/bin/gcovr \
--root "$GITHUB_WORKSPACE" \
--exclude '^build/' \
--exclude '^tests/' \
--exclude-throw-branches \
--gcov-ignore-parse-errors=negative_hits.warn \
--xml coverage.xml \
--txt-summary \
"$GITHUB_WORKSPACE/build"
# Fail here instead of silently uploading an empty report to Codecov
grep -q '<class ' coverage.xml || { echo "::error::coverage.xml contains no files"; exit 1; }

- name: Upload logs artifact
if: always()
- name: Upload coverage artifact
if: matrix.build_type == 'Debug' && matrix.compiler == 'gcc:16'
uses: actions/upload-artifact@v7
with:
name: capiocl-logs-${{ matrix.build_type }}-${{ strategy.job-index }}
path: capiocl_logs.zip
name: coverage
path: coverage.xml
retention-days: 1
if-no-files-found: ignore

upload-to-codecov:
name: "compute Codecov"
needs: [ "unit-tests"]
runs-on: ubuntu-latest
env:
CALF_LOG_DIR: ${{ github.workspace }}/capiocl_logs
steps:
- uses: actions/checkout@v7

- name: "Install packages"
run: |
sudo apt-get update
sudo apt-get install -y git \
cmake make ninja-build sudo \
python3 python3-pip python3-venv \
gcc g++ libcurl4-gnutls-dev libprotobuf-dev protobuf-compiler

- name: "Compile tests"
run: |
cmake -DCMAKE_BUILD_TYPE=Debug \
-DENABLE_COVERAGE=ON \
-DBUILD_PYTHON_BINDINGS=OFF \
-DCAPIO_CL_BUILD_TESTS=ON \
-G Ninja \
-B ../build \
-S ${GITHUB_WORKSPACE}

cmake --build ../build -j $(nproc)
sudo cmake --install ../build --prefix /usr/local

- name: Run unit tests
shell: bash
run: CAPIO_CL_tests --gtest_break_on_failure --gtest_print_time=1
if-no-files-found: error

- name: Zip logs
if: always()
shell: bash
run: |
if [ -d "capiocl_logs" ]; then
cmake -E tar cf capiocl_logs.zip --format=zip capiocl_logs
Expand All @@ -126,35 +111,29 @@ jobs:
if: always()
uses: actions/upload-artifact@v7
with:
name: capiocl-logs-coverage
name: capiocl-logs-${{ matrix.build_type }}-${{ strategy.job-index }}
path: capiocl_logs.zip
retention-days: 1
if-no-files-found: ignore

- name: "Generate coverage report"
run: |
pip install --upgrade gcovr
gcovr \
--exclude-throw-branches \
--gcov-ignore-parse-errors=negative_hits.warn \
--exclude tests/ \
--xml coverage.xml \
../build
upload-to-codecov:
name: "compute Codecov"
needs: [ "unit-tests" ]
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7

- name: Upload coverage artifact
uses: actions/upload-artifact@v7
- name: "Download coverage artifact"
uses: actions/download-artifact@v8
with:
name: coverage
path: coverage.xml
retention-days: 1

- name: "Download artifacts"
uses: actions/download-artifact@v8

- name: "Upload coverage to Codecov"
uses: codecov/codecov-action@v7
with:
fail_ci_if_error: true
files: ./coverage.xml
disable_search: true
token: ${{ secrets.CODECOV_TOKEN }}

build-documentation:
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/python-bindings.yml
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ jobs:
name: "Test Python bindings"
strategy:
matrix:
on: [ 'ubuntu-24.04', 'macos-15-intel', 'macos-15' ]
on: [ 'ubuntu-24.04', 'macos-15' ]
python: [ '3.10', '3.11', '3.12', '3.13', '3.14' ]

runs-on: ${{ matrix.on }}
Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
| OS / Arch | ![x86_64](https://img.shields.io/badge/x86__64-121212?logo=intel&logoColor=blue) | ![ARM](https://img.shields.io/badge/ARM-121212?logo=arm&logoColor=0091BD) | ![RISC-V](https://img.shields.io/badge/RISC--V-121212?logo=riscv&logoColor=F9A825) |
|------------------------------------------------------------------------------------|----------------------------------------------------------------------------------|---------------------------------------------------------------------------|------------------------------------------------------------------------------------|
| ![Ubuntu](https://img.shields.io/badge/Ubuntu-121212?logo=ubuntu&logoColor=E95420) | YES | YES | YES |
| ![macOS](https://img.shields.io/badge/macOS-121212?logo=apple&logoColor=white) | YES | YES | N.A. |
| ![macOS](https://img.shields.io/badge/macOS-121212?logo=apple&logoColor=white) | No CI/CD support | YES | N.A. |

#### Documentation

Expand Down
1 change: 1 addition & 0 deletions bindings/python_bindings.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ PYBIND11_MODULE(_py_capio_cl, m) {
.def("setWorkflowName", &capiocl::engine::Engine::setWorkflowName, py::arg("name"))
.def("setCommitted", &capiocl::engine::Engine::setCommitted, py::arg("path"))
.def("isCommitted", &capiocl::engine::Engine::isCommitted, py::arg("path"))
.def("increaseCloseCount", &capiocl::engine::Engine::increaseCloseCount, py::arg("path"))
.def("setHomeNode", &capiocl::engine::Engine::setHomeNode, py::arg("path"))
.def("getPaths", &capiocl::engine::Engine::getPaths)
.def("startApiServer", &capiocl::engine::Engine::startApiServer)
Expand Down
11 changes: 10 additions & 1 deletion capiocl/engine.h
Original file line number Diff line number Diff line change
Expand Up @@ -427,6 +427,15 @@ class Engine final {
*/
bool isCommitted(const std::filesystem::path &path) const;

/**
* @brief Record one qualified logical close for an ON_CLOSE path.
* Filesystem-backed counts greater than one require CAPIO_METADATA_DIR to identify a trusted,
* unique-per-run metadata directory with coherent atomic exclusive create/unlink. Multicast
* counts are volatile and eventual.
* @return true when the path is effectively committed after this call.
*/
bool increaseCloseCount(const std::filesystem::path &path) const;

/**
* Set file indicated by path as committed
* @param path
Expand Down Expand Up @@ -466,4 +475,4 @@ class Engine final {

} // namespace capiocl::engine

#endif // CAPIO_CL_ENGINE_H
#endif // CAPIO_CL_ENGINE_H
37 changes: 31 additions & 6 deletions capiocl/monitor.h
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@

#include <atomic>
#include <cstddef>
#include <cstdint>
#include <filesystem>
#include <mutex>
#include <optional>
#include <set>
#include <string>
#include <thread>
Expand Down Expand Up @@ -106,6 +108,13 @@ class MonitorInterface {
*/
virtual void setCommitted(const std::filesystem::path &path) const;

/**
* @brief Atomically record a close and commit when @p threshold is reached.
* @return the committed state, or std::nullopt when this backend has no persistent counter.
*/
virtual std::optional<bool> increaseCloseCount(const std::filesystem::path &path,
long threshold) const;

/**
* Set the current hostname as the home node for a given path
* @param path
Expand Down Expand Up @@ -149,6 +158,12 @@ class MulticastMonitor final : public MonitorInterface {
/// @brief variable to terminate execution
std::atomic<bool> terminate = false;

/// Per-path maximum close-count snapshot observed from each origin.
mutable std::unordered_map<std::string, std::unordered_map<std::string, std::uint64_t>>
close_counts;
mutable std::mutex close_count_lock;
std::string close_count_origin;

/// @brief Multicast poll timeout interval
static constexpr int MULTICAST_THREAD_POLL_INTERVAL = 250;

Expand All @@ -165,17 +180,17 @@ class MulticastMonitor final : public MonitorInterface {
/**
* @brief Supported network command types for commit messages.
*/
typedef enum { SET = '!', GET = '?' } MESSAGE_COMMANDS;
typedef enum { SET = '!', GET = '?', COUNT_SET = 'C', COUNT_GET = 'Q' } MESSAGE_COMMANDS;

/**
* @brief Send a commit or request message over multicast.
*
* @param ip_addr Destination multicast address.
* @param ip_port Destination multicast port.
* @param path File path associated with the message.
* @param payload Message payload.
* @param action The type of message to send (COMMIT or REQUEST).
*/
static void _send_message(const std::string &ip_addr, int ip_port, const std::string &path,
static void _send_message(const std::string &ip_addr, int ip_port, const std::string &payload,
MESSAGE_COMMANDS action);

/**
Expand All @@ -191,9 +206,12 @@ class MulticastMonitor final : public MonitorInterface {
* @param ip_port Multicast commit listen port.
* @param terminate Atomic Boolean flag to terminate thread
*/
static void commit_listener(std::vector<std::string> &committed_files, std::mutex &lock,
const std::string &ip_addr, int ip_port,
const std::atomic<bool> *terminate);
static void
commit_listener(std::vector<std::string> &committed_files, std::mutex &lock,
std::unordered_map<std::string, std::unordered_map<std::string, std::uint64_t>>
&close_counts,
std::mutex &close_count_lock, const std::string &ip_addr, int ip_port,
const std::atomic<bool> *terminate);

/**
* @brief Background thread function to listen for commit messages.
Expand Down Expand Up @@ -227,6 +245,8 @@ class MulticastMonitor final : public MonitorInterface {

bool isCommitted(const std::filesystem::path &path) const override;
void setCommitted(const std::filesystem::path &path) const override;
std::optional<bool> increaseCloseCount(const std::filesystem::path &path,
long threshold) const override;
void setHomeNode(const std::filesystem::path &path) const override;
std::string getHomeNode(const std::filesystem::path &path) const override;
};
Expand Down Expand Up @@ -286,6 +306,8 @@ class FileSystemMonitor final : public MonitorInterface {

bool isCommitted(const std::filesystem::path &path) const override;
void setCommitted(const std::filesystem::path &path) const override;
std::optional<bool> increaseCloseCount(const std::filesystem::path &path,
long threshold) const override;
void setHomeNode(const std::filesystem::path &path) const override;
std::string getHomeNode(const std::filesystem::path &path) const override;
};
Expand Down Expand Up @@ -319,6 +341,9 @@ class Monitor {
*/
void setCommitted(std::filesystem::path path) const;

/// @brief Record a close independently in every counter-capable backend.
[[nodiscard]] bool increaseCloseCount(const std::filesystem::path &path, long threshold) const;

/**
* Add a new backend for monitor. Must be a derived class from MonitorInterface
* @param interface
Expand Down
17 changes: 15 additions & 2 deletions doxygen/semantics.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,18 @@ commit behaviors:

- `Commit on Close (CoC)`: This behavior allows subsequent steps to initiate reading a file as soon as the producer step
invokes a close operation on a given file, signaling that all I/O operations on that file are completed. It is also
allowed to consider a file committed after `n` close operations are performed. The following Gant diagram visually
allowed to consider a file committed after `n` close operations are performed. Counts of zero or one commit on the
first close; larger counts require a counter-capable filesystem or multicast monitor. Filesystem counters require
`CAPIO_METADATA_DIR` to name a trusted, non-attacker-writable directory unique to the workflow run. CAPIO-CL owns its
`capiocl` subtree. Multi-process and multi-node producers must share this metadata directory through a distributed
filesystem providing coherent atomic exclusive file creation and unlink.
Transient locks use a private subtree separate from counters. Counter replacement is atomic, but CAPIO-CL does not
claim power-loss durability for the latest close on distributed storage.
A producer crash can leave a stale lock file; CAPIO-CL does not steal or expire such locks automatically.
Multicast counters exchange monotonic per-origin snapshots and are eventual and volatile. UDP loss can defer
commitment, and origin death before propagation can lose a close; there is no durability, consensus, or
authentication.
The following Gant diagram visually
explains the CoC semantics:

![The Commit on Close rule](media/coc.png){ width=60% }
Expand All @@ -33,7 +44,9 @@ commit behaviors:
when the number of `open()` and `close()` system calls operations for a given file is not statically known. Instead,
we are aware that the I/O operations can be considered concluded on a given file if another file has been committed.
This additional commit behavior introduces a dependency among files in the commit rule, expanding opportunities to
leverage temporal parallelism for I/O operations across different workflow steps.The following Gant diagram visually
leverage temporal parallelism for I/O operations across different workflow steps. Multiple dependencies use AND
semantics and are evaluated transitively; unresolved dependency cycles remain uncommitted. Runtime dependencies are
concrete literal paths, and dependency glob patterns do not resolve. The following Gant diagram visually
explains the CoF semantics:

![The Commit on File rule](media/cof.png){ width=60% }
Expand Down
Loading
Loading