dlang/cdc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

init

Browse source
This commit is contained in:
Alexander Zhirov 2025-09-04 21:51:23 +03:00
commit f1dd2aebb2
Signed by: alexander
GPG key ID: C8D8BE544A27C511
8 changed files with 642 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

17
.gitignore vendored Normal file
View file

@ -0,0 +1,17 @@
.dub
docs.json
__dummy.html
docs/
/cdc
cdc.so
cdc.dylib
cdc.dll
cdc.a
cdc.lib
cdc-test-*
*.exe
*.pdb
*.o
*.obj
*.lst
bin

39
.vscode/launch.json vendored Normal file
View file

@ -0,0 +1,39 @@
{
// Используйте IntelliSense, чтобы узнать о возможных атрибутах.
// Наведите указатель мыши, чтобы просмотреть описания существующих атрибутов.
// Для получения дополнительной информации посетите: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"type": "code-d",
"request": "launch",
"dubBuild": true,
"name": "Build & Debug DUB project",
"cwd": "${command:dubWorkingDirectory}",
"program": "bin/${command:dubTarget}",
"args": [
"/tmp/test"
]
},
{
"name": "Debug D Program with sudo-gdb",
"type": "cppdbg",
"request": "launch",
"program": "${workspaceFolder}/bin/dwatch",
"args": ["-d", "/tmp/scripts"], // Аргументы командной строки для программы, если нужны
"stopAtEntry": false, // Остановить на входе в main
"cwd": "${workspaceFolder}",
"environment": [],
"externalConsole": false,
"MIMode": "gdb",
"miDebuggerPath": "/usr/bin/sudo-gdb", // Путь к вашему скрипту
"setupCommands": [
{
"description": "Enable pretty-printing for gdb",
"text": "-enable-pretty-printing",
"ignoreFailures": true
}
]
}
]
}

5
.vscode/settings.json vendored Normal file
View file

@ -0,0 +1,5 @@
{
"editor.insertSpaces": false,
"editor.tabSize": 4,
"editor.detectIndentation": false
}

1
README.md Normal file
View file

@ -0,0 +1 @@
# Implementing Change Data Capture

11
dub.json Normal file
View file

@ -0,0 +1,11 @@
{
"authors": [
"Alexander Zhirov"
],
"copyright": "Copyright © 2025, Alexander Zhirov",
"description": "Implementing Change Data Capture.",
"license": "BSL-1.0",
"name": "cdc",
"targetPath": "bin",
"targetType": "executable"
}

5
dub.selections.json Normal file
View file

@ -0,0 +1,5 @@
{
"fileVersion": 1,
"versions": {
}
}

272
source/app.d Normal file
View file

@ -0,0 +1,272 @@
module app;
import std.stdio : writeln, writefln, File;
import std.file : exists, mkdirRecurse, read, write, readText;
import std.path : baseName, buildPath, absolutePath;
import std.getopt : getopt;
import std.string : strip, split, splitLines;
import std.algorithm.searching : startsWith;
import std.conv : to;
import std.datetime : Clock;
import std.exception : enforce;
import std.digest.sha : sha256Of;
import fastcdc; // твой модуль FastCDC
// ---------- утилиты ----------
// hex: параметр scope, чтобы можно было безопасно передавать срез локального массива
@safe pure
string toHex(scope const(ubyte)[] bytes)
{
immutable char[16] HEX = "0123456789abcdef";
auto buf = new char[bytes.length * 2];
size_t j = 0;
foreach (b; bytes)
{
buf[j++] = HEX[(b >> 4) & 0xF];
buf[j++] = HEX[b & 0xF];
}
return buf.idup; // immutable string
}
// fanout: store/chunks/aa/bb/<hash>.bin
@safe
string chunkPath(string storeDir, string hashHex)
{
auto a = hashHex[0 .. 2];
auto b = hashHex[2 .. 4];
return buildPath(storeDir, "chunks", a, b, hashHex ~ ".bin");
}
// manifest: store/manifests/<name>.<epoch>.manifest
@safe
string manifestPath(string storeDir, string srcPath, long epoch)
{
auto name = baseName(srcPath);
return buildPath(storeDir, "manifests", name ~ "." ~ to!string(epoch) ~ ".manifest");
}
@safe
void ensureDirs(string storeDir)
{
mkdirRecurse(buildPath(storeDir, "chunks"));
mkdirRecurse(buildPath(storeDir, "manifests"));
}
@trusted ubyte[] readBytes(string path)
{
// std.file.read выделяет новый буфер байтов → безопасно привести к ubyte[]
auto v = read(path); // void[]
return cast(ubyte[]) v; // доверяем Phobos: это сырой байтовый буфер
}
// ---------- split ----------
struct SplitOpts
{
string storeDir;
string filePath;
size_t minSize = 8 * 1024;
size_t avgSize = 64 * 1024;
size_t maxSize = 256 * 1024;
}
@safe
int cmdSplit(SplitOpts opt)
{
enforce(exists(opt.filePath), "Файл не найден: " ~ opt.filePath);
ensureDirs(opt.storeDir);
// бинарное чтение: std.file.read возвращает ubyte[]
ubyte[] data = readBytes(opt.filePath);
FastCDCParams p = {opt.minSize, opt.avgSize, opt.maxSize};
p.normalize();
size_t chunkCount = 0;
size_t totalBytes = data.length;
auto epoch = Clock.currTime().toUnixTime();
auto mfPath = manifestPath(opt.storeDir, opt.filePath, epoch);
mkdirRecurse(buildPath(opt.storeDir, "manifests"));
auto mf = File(mfPath, "w");
// шапка манифеста
mf.writeln("# FastCDC manifest");
mf.writefln("path\t%s", absolutePath(opt.filePath));
mf.writefln("size\t%s", to!string(totalBytes));
mf.writefln("algo\tsha256");
mf.writefln("min\t%u", cast(uint) p.minSize);
mf.writefln("avg\t%u", cast(uint) p.avgSize);
mf.writefln("max\t%u", cast(uint) p.maxSize);
mf.writeln("ord\thash\tsize");
size_t ord = 0;
processStream(data, p, (size_t start, size_t len) @safe {
auto slice = data[start .. start + len];
auto digest = sha256Of(slice); // ubyte[32] (на стеке)
auto hex = toHex(digest[]); // scope-параметр — ок
mkdirRecurse(buildPath(opt.storeDir, "chunks", hex[0 .. 2]));
mkdirRecurse(buildPath(opt.storeDir, "chunks", hex[0 .. 2], hex[2 .. 4]));
auto cpath = chunkPath(opt.storeDir, hex);
if (!exists(cpath))
write(cpath, slice);
mf.writefln("%u\t%s\t%u", cast(uint) ord, hex, cast(uint) len);
++ord;
++chunkCount;
return 0;
});
mf.flush();
mf.close();
writefln("split: %s", opt.filePath);
writefln("store: %s", opt.storeDir);
writefln("manifest: %s", mfPath);
writefln("chunks: %u, bytes: %u",
cast(uint) chunkCount, cast(uint) totalBytes);
return 0;
}
// ---------- restore ----------
struct RestoreOpts
{
string storeDir;
string manifestFile;
string outFile;
}
@safe
int cmdRestore(RestoreOpts opt)
{
enforce(exists(opt.manifestFile), "Манифест не найден: " ~ opt.manifestFile);
string text = readText(opt.manifestFile);
auto lines = splitLines(text);
// найти строку "ord\thash\tsize"
size_t i = 0;
while (i < lines.length && !lines[i].strip.startsWith("ord"))
++i;
enforce(i < lines.length, "Не найден заголовок секции данных в манифесте");
++i;
auto dst = File(opt.outFile, "wb");
size_t count = 0;
for (; i < lines.length; ++i)
{
auto ln = lines[i].strip;
if (ln.length == 0 || ln[0] == '#')
continue;
auto cols = ln.split('\t');
enforce(cols.length == 3, "Строка манифеста повреждена: " ~ ln);
auto hashHex = cols[1];
auto cpath = chunkPath(opt.storeDir, hashHex);
enforce(exists(cpath), "Чанк не найден: " ~ cpath);
ubyte[] chunkData = readBytes(cpath);
dst.rawWrite(chunkData);
++count;
}
dst.close();
writefln("restore: %s <- %s (chunks: %u)",
opt.outFile, opt.manifestFile, cast(uint) count);
return 0;
}
// ---------- CLI ----------
@safe
void printHelp(string prog)
{
writeln("Usage:");
writeln(" ", prog, " split --store <dir> <file> [--min N] [--avg N] [--max N]");
writeln(" ", prog, " restore --store <dir> <manifest> <out_file>");
}
int main(string[] args) // без @safe: getopt требует &var
{
if (args.length < 2)
{
printHelp(args[0]);
return 1;
}
switch (args[1])
{
case "split":
{
SplitOpts opt;
string store;
size_t minS = 0, avgS = 0, maxS = 0;
auto res = getopt(args,
"store", &store,
"min", &minS,
"avg", &avgS,
"max", &maxS
);
if (res.helpWanted)
{
printHelp(args[0]);
return 0;
}
// после getopt в args остаются позиционные
if (args.length < 3 || store.length == 0)
{
printHelp(args[0]);
return 1;
}
opt.storeDir = store;
opt.filePath = args[2];
if (minS)
opt.minSize = minS;
if (avgS)
opt.avgSize = avgS;
if (maxS)
opt.maxSize = maxS;
return cmdSplit(opt);
}
case "restore":
{
RestoreOpts opt;
string store;
auto res = getopt(args, "store", &store);
if (res.helpWanted)
{
printHelp(args[0]);
return 0;
}
if (args.length < 4 || store.length == 0)
{
printHelp(args[0]);
return 1;
}
opt.storeDir = store;
opt.manifestFile = args[2];
opt.outFile = args[3];
return cmdRestore(opt);
}
default:
printHelp(args[0]);
return 1;
}
}

292
source/fastcdc.d Normal file
View file

@ -0,0 +1,292 @@
module fastcdc;
/**
* FastCDC (скользящее контент-зависимое разбиение).
*
* Идея: идём по данным байт за байтом, считаем rolling-hash (Gear),
* и ставим границы чанков там, где hash удовлетворяет простому условию:
* hash & mask == 0
*
* При этом действуют три порога:
* - minSize: до этой длины чанк не разрываем (слишком мелко)
* - avgSize: «целевой» средний размер; после него границы ищем активнее
* - maxSize: жёсткий максимум; если так и не нашли границу — режем принудительно
*
* Благодаря контент-зависимым границам, маленькие вставки/удаления
* меняют только локальные чанки; большинство остальных переиспользуется.
*/
/// Параметры FastCDC. Рекомендуется avgSize как степень двойки (например, 64 KiB).
struct FastCDCParams
{
/// Минимальная длина чанка: пока длина меньше — границы не ставим.
size_t minSize = 8 * 1024;
/// Целевой средний размер чанка (желательно 2^k: 32, 64, 128 KiB и т.п.).
size_t avgSize = 64 * 1024;
/// Жёсткий максимум длины чанка: достигли — режем независимо от хэша.
size_t maxSize = 256 * 1024;
/**
* Нормализация границ (делает параметры самосогласованными):
* - minSize >= 4 KiB
* - avgSize > minSize (если нет — удваиваем)
* - maxSize > avgSize (если нет — умножаем на 4)
*
* @safe nothrow @nogc — чтобы можно было вызывать из безопасных функций
* без аллокаций/исключений.
*/
@safe nothrow @nogc
void normalize()
{
enum ki = 1024;
if (minSize < 4 * ki)
minSize = 4 * ki;
if (avgSize <= minSize)
avgSize = minSize * 2;
if (maxSize <= avgSize)
maxSize = avgSize * 4;
}
}
/// Диапазон чанка: [start .. start+len) в исходном буфере.
struct ChunkRange
{
size_t start; // смещение начала чанка
size_t len; // длина чанка в байтах
}
/**
* Таблица для Gear-rolling-hash: 256 псевдослучайных 32-битных констант.
* Хэш обновляется так: h = (h << 1) + GEAR[byte].
* Константы фиксированы → детерминированные границы при одинаковых параметрах.
*/
private immutable uint[256] GEAR = [
0x4f1bbcdc, 0xe47c2b1a, 0x1a16b407, 0xa88f9d43, 0x33b0b5c5, 0x0f0b1192,
0xb1c2e5b8, 0xc6f5a2a9,
0x7d1ea04c, 0x26358f23, 0x8f9a7902, 0x5ab7d6ee, 0x2f6d3a8c, 0x9e13c540,
0x4d7c8b99, 0xf3a1d2b1,
0x0b7d4c62, 0x81f2a5d3, 0x19a8b604, 0x6cc7e190, 0x559e43a1, 0xd2f01937,
0x7a53cc4f, 0x0c1d5e66,
0x3b7f6a22, 0x99d104b7, 0x4aa7f9e1, 0x2ce3b8c0, 0x6a8f73d4, 0xe1b2c9a8,
0x57c04b13, 0xa4e91572,
0x13f7c2aa, 0x8b1c0f5e, 0x5e6a92c1, 0x0af41937, 0x7fe0bd54, 0x26b3e71a,
0x942d6c83, 0x3c51a0ef,
0xd57f2b33, 0x61a4cc09, 0x0d9b8a71, 0xb7e50f46, 0x48a3d1f0, 0x2f1e6cb2,
0x73cd98a5, 0xe92a13c9,
0xa1c7f02e, 0x5b0e6a97, 0x0c8f2d31, 0xd1a47b66, 0x6fe3b920, 0x20b9d4a1,
0x9a5c0f3d, 0x4e81a2c7,
0xf02b5934, 0x1bc7d8a2, 0x8e0a64f1, 0x37d4b20c, 0x6c09f5d3, 0xa2391e84,
0x5f7ab0e2, 0x0b1c6d57,
0x7c3f9a15, 0x12ad54e3, 0x8b6f0c2d, 0x45e1d7a9, 0x2af39b60, 0x9c07e4d1,
0x3d5a81b2, 0xe6c21458,
0xd9a03f1c, 0x64b7e0a3, 0x0ea19c76, 0xb2d5480f, 0x49f3a7d5, 0x21c58e92,
0x75ae39c1, 0xed1046ab,
0xa8c3f12d, 0x5c0e7b94, 0x0d8f2e31, 0xd4a57c66, 0x6ee4ba20, 0x23bad5a1,
0x985d0f3d, 0x4f82a3c7,
0xf12c5a34, 0x1ac8d9a2, 0x8f0b65f1, 0x36d5b30c, 0x6b0af6d3, 0xa33a1f84,
0x5e7bb1e2, 0x0a1d6e57,
0x7d409b15, 0x11ae55e3, 0x8a700d2d, 0x44e2d8a9, 0x2bf49c60, 0x9d08e5d1,
0x3c5b82b2, 0xe7c31558,
0xd8a1401c, 0x65b8e1a3, 0x0fa29d76, 0xb3d6490f, 0x48f4a8d5, 0x22c68f92,
0x74af3ac1, 0xec1147ab,
0xa9c4f22d, 0x5d0f7c94, 0x0e902f31, 0xd5a67d66, 0x6de5bb20, 0x24bbd6a1,
0x975e103d, 0x4e83a4c7,
0xf22d5b34, 0x1bc9daa2, 0x8e0c66f1, 0x35d6b40c, 0x6a0bf7d3, 0xa23b2084,
0x5f7cb2e2, 0x0b1e6f57,
0x7c419c15, 0x10af56e3, 0x8b710e2d, 0x45e3d9a9, 0x2af59d60, 0x9c09e6d1,
0x3d5c83b2, 0xe6c41658,
0xd9a2411c, 0x64b9e2a3, 0x0ea39e76, 0xb2d74a0f, 0x49f5a9d5, 0x23c79092,
0x75b03bc1, 0xed1248ab,
0xa8c5f32d, 0x5c107d94, 0x0d913031, 0xd4a77e66, 0x6ee6bc20, 0x25bcd7a1,
0x985f113d, 0x4f84a5c7,
0xf12e5c34, 0x1acadba2, 0x8f0d67f1, 0x36d7b50c, 0x6b0cf8d3, 0xa33c2184,
0x5e7db3e2, 0x0a1f7057,
0x7d429d15, 0x11b057e3, 0x8a720f2d, 0x44e4daa9, 0x2bf69e60, 0x9d0ae7d1,
0x3c5d84b2, 0xe7c51758,
0xd8a3421c, 0x65bae3a3, 0x0fa49f76, 0xb3d84b0f, 0x48f6aad5, 0x22c89192,
0x74b13cc1, 0xec1349ab,
0xa9c6f42d, 0x5d117e94, 0x0e923131, 0xd5a87f66, 0x6de7bd20, 0x24bdd8a1,
0x9750123d, 0x4e85a6c7,
0xf22f5d34, 0x1bcbdca2, 0x8e0e68f1, 0x35d8b60c, 0x6a0df9d3, 0xa23d2284,
0x5f7eb4e2, 0x0b207157,
0x7c439e15, 0x10b158e3, 0x8b73102d, 0x45e5dba9, 0x2af79f60, 0x9c0be8d1,
0x3d5e85b2, 0xe6c61858,
0xd9a4431c, 0x64bbe4a3, 0x0ea5a076, 0xb2d94c0f, 0x49f7abd5, 0x23c99292,
0x75b23dc1, 0xed144aab,
0xa8c7f52d, 0x5c127f94, 0x0d933231, 0xd4a98066, 0x6ee8be20, 0x25bed9a1,
0x9861133d, 0x4f86a7c7,
0xf1305e34, 0x1acca0a2, 0x8f0f69f1, 0x36d9b70c, 0x6b0efaD3, 0xa33e2384,
0x5e7fb5e2, 0x0a217257,
0x7d449f15, 0x11b259e3, 0x8a74112d, 0x44e6dca9, 0x2bf8a060, 0x9d0ce9d1,
0x3c5f86b2, 0xe7c71958,
0xd8a5441c, 0x65bce5a3, 0x0fa6a176, 0xb3da4d0f, 0x48f8acd5, 0x22ca9392,
0x74b33ec1, 0xec154bab
];
/// Обновление Gear-хэша одним байтом. @safe @nothrow @nogc — без аллокаций/исключений.
@safe nothrow @nogc
private uint gearUpdate(uint h, ubyte b)
{
// Побитовый сдвиг + добавление таблицы → быстрый, но «живой» rolling-hash.
return (h << 1) + GEAR[b];
}
/**
* Ближайшая вниз степень двойки для x (>=1).
* Нужна, чтобы построить маски вида (2^k − 1) и (2^(k+1) − 1).
* При таких масках условие (hash & mask) == 0 имеет ожидаемый период ~2^k.
*/
@safe nothrow @nogc
private size_t floorPow2(size_t x)
{
size_t p = 1;
while ((p << 1) != 0 && (p << 1) <= x)
p <<= 1;
return p;
}
/**
* Batch-API: режет весь буфер и возвращает список чанков.
* Удобно, если объём данных умеренный и важна простота.
*
* Сложность: O(n), где n — длина data.
*/
@safe
ChunkRange[] chunkify(const(ubyte)[] data, FastCDCParams params = FastCDCParams.init)
{
FastCDCParams p = params;
p.normalize(); // приводим пороги в адекватные границы
ChunkRange[] chunks; // итоговый список диапазонов
// Необязательный pre-reserve: уменьшает реаллокации; можно удалить, если не нужно.
chunks.reserve(data.length / (p.avgSize ? p.avgSize : 1) + 1);
// Строим две маски:
// - maskN (normal): 2^k − 1 → «частота» границ ~ avgSize
// - maskE (early): 2^(k+1) − 1 → «строже», реже до avgSize (не режем слишком рано)
const size_t avgPow2 = floorPow2(p.avgSize);
immutable uint maskN = cast(uint)(avgPow2 - 1);
immutable uint maskE = cast(uint)(((avgPow2 << 1) != 0) ? ((avgPow2 << 1) - 1) : maskN);
size_t start = 0; // начало текущего потенциального чанка
uint rolling = 0; // текущее значение rolling-hash
foreach (i, b; data) // идём по байтам
{
rolling = gearUpdate(rolling, b); // обновили rolling-hash
const size_t clen = (i + 1) - start; // текущая длина кандидат-чанка
// 1) До minSize границы запрещены — продолжаем копить байты.
if (clen < p.minSize)
continue;
// 2) Если превысили maxSize — принудительно режем тут.
if (clen >= p.maxSize)
{
chunks ~= ChunkRange(start, clen);
start = i + 1; // новый чанк начинается со следующего байта
rolling = 0; // сбрасываем хэш (независимая последовательность)
continue;
}
// 3) Зона «ранняя» (min..avg): применяем более строгую маску, чтобы
// не нарезать слишком короткие чанки.
if (clen < p.avgSize)
{
if ((rolling & maskE) == 0) // редкое совпадение → ставим границу
{
chunks ~= ChunkRange(start, clen);
start = i + 1;
rolling = 0;
}
continue; // иначе продолжаем наращивать чанк
}
// 4) Зона «нормальная» (avg..max): используем обычную маску — средний размер ≈ avgSize.
if ((rolling & maskN) == 0)
{
chunks ~= ChunkRange(start, clen);
start = i + 1;
rolling = 0;
}
}
// 5) Хвост: если что-то осталось — это последний чанк.
if (start < data.length)
chunks ~= ChunkRange(start, data.length - start);
return chunks;
}
/**
* Стримовый API: вызывает sink(start,len) для каждого найденного чанка.
*
* Плюсы:
* - не аллоцирует список чанков (можно сделать @nogc-путь);
* - удобно сразу считать хэш чанка и писать в БД.
*
* Контракт sink:
* - возвращает 0 → «продолжать»;
* - возвращает != 0 → «остановить» (функция вернёт количество уже эмитнутых чанков).
*/
@safe
size_t processStream(const(ubyte)[] data, FastCDCParams params,
scope int delegate(size_t, size_t) @safe sink)
{
FastCDCParams p = params;
p.normalize();
const size_t avgPow2 = floorPow2(p.avgSize);
immutable uint maskN = cast(uint)(avgPow2 - 1);
immutable uint maskE = cast(uint)(((avgPow2 << 1) != 0) ? ((avgPow2 << 1) - 1) : maskN);
size_t start = 0;
uint rolling = 0;
size_t count = 0; // числим эмитнутые чанки
foreach (i, b; data)
{
rolling = gearUpdate(rolling, b);
const size_t clen = (i + 1) - start;
// До minSize границы не допускаются
if (clen < p.minSize)
continue;
// Решение «резать/не резать» для текущей позиции
bool cut = false;
if (clen >= p.maxSize) // принудительный разрыв на maxSize
cut = true;
else if (clen < p.avgSize) // раннее окно — строгая маска
cut = ((rolling & maskE) == 0);
else // нормальное окно — обычная маска
cut = ((rolling & maskN) == 0);
if (cut)
{
// Отдаём чанк потребителю. Он может попросить остановиться (!=0).
if (sink(start, clen) != 0)
{
++count; // этот чанк уже отдан — учитываем
return count; // и выходим
}
++count;
start = i + 1; // следующий чанк начинается со следующего байта
rolling = 0; // сбрасываем rolling-hash
}
}
// Хвостовой чанк (если есть): отдаём целиком.
if (start < data.length)
{
if (sink(start, data.length - start) != 0)
{
++count; // учли последний чанк
return count; // останов по сигналу sink
}
++count;
}
return count;
}