diff --git a/Makefile b/Makefile index 5e1905b24..f6467dba2 100644 --- a/Makefile +++ b/Makefile @@ -26,27 +26,28 @@ include ports.mk PWD := $(shell pwd) BUILD ?= build ARCH ?= $(HOSTARCH) -OBJ := $(BUILD)/$(ARCH)/obj -BIN := $(BUILD)/$(ARCH)/bin -LIB := $(BUILD)/$(ARCH)/lib -TESTS := $(BUILD)/$(ARCH)/tests -TMPBIN := $(BUILD)/$(ARCH)/tmp -INC := $(BUILD)/$(ARCH)/include +OBJ := $(BUILD)/$(ARCH)-$(OSNAME)/obj +BIN := $(BUILD)/$(ARCH)-$(OSNAME)/bin +LIB := $(BUILD)/$(ARCH)-$(OSNAME)/lib +TESTS := $(BUILD)/$(ARCH)-$(OSNAME)/tests +TMPBIN := $(BUILD)/$(ARCH)-$(OSNAME)/tmp +INC := $(BUILD)/$(ARCH)-$(OSNAME)/include SRC := . -TMP ?= $(PWD)/$(BUILD)/$(ARCH)/tmp +TMP ?= $(PWD)/$(BUILD)/$(ARCH)-$(OSNAME)/tmp # These are for cross-compilation, where binaries used in the build need # be be built for the host. HOSTARCH ?= $(ARCH) -HOSTBIN ?= $(BUILD)/$(HOSTARCH)/bin -HOSTLIB ?= $(BUILD)/$(HOSTARCH)/lib -HOSTINC ?= $(BUILD)/$(HOSTARCH)/include +HOSTOSNAME ?= $(OSNAME) +HOSTBIN ?= $(BUILD)/$(HOSTARCH)-$(HOSTOSNAME)/bin +HOSTLIB ?= $(BUILD)/$(HOSTARCH)-$(HOSTOSNAME)/lib +HOSTINC ?= $(BUILD)/$(HOSTARCH)-$(HOSTOSNAME)/include TEST_TMP := $(TESTS) # Vars for configuration files or files that live outside bin and lib ALTROOT := $(BUILD)/$(ARCH)/altroot ETC := $(ALTROOT)/etc -PLUGINS := $(BUILD)/$(ARCH)/mldb_plugins +PLUGINS := $(BUILD)/$(ARCH)-$(OSNAME)/mldb_plugins JML_BUILD := mldb/jml-build INCLUDE := -Imldb diff --git a/block/content_descriptor.cc b/block/content_descriptor.cc index 7ecd5ed45..ab4ef7cf4 100644 --- a/block/content_descriptor.cc +++ b/block/content_descriptor.cc @@ -460,13 +460,22 @@ getStream(const std::map & options) const //cerr << "url = " << descriptor.getUrlStringUtf8() << " compression = " // << compression << " mapped = " << isMapped << endl; - if (isMapped) { + while (isMapped) { // actually an if, but now we can break out // Just get one single big block auto contentHandler = getContent(descriptor); struct Vals { FsObjectInfo info; FrozenMemoryRegion mem; + +#if 0 + ~Vals() + { + cerr << endl << endl << endl; + cerr << "NO MORE MAPPING VALS" << endl; + cerr << endl << endl << endl; + } +#endif }; auto vals = std::make_shared(); @@ -493,7 +502,10 @@ getStream(const std::map & options) const vals->mem.length()); if (outputSize < 0) { - throw Exception("decompressed size unknown"); + if (outputSize == Decompressor::LENGTH_UNKNOWN) + break; // do as an istream as we can't run the splitting algorithm + + throw Exception("decompressed size unknown: %i", (int)outputSize); } static MemorySerializer serializer; @@ -546,10 +558,8 @@ getStream(const std::map & options) const filter_istream stream(handler, descriptor.getUrlStringUtf8(), options2); return stream; } - else { - // Not mapped. We go block by block. - } - + + // Not mapped. We go block by block. filter_istream result(descriptor.getUrlStringUtf8(), options2); return result; } diff --git a/block/testing/content_descriptor_test.cc b/block/testing/content_descriptor_test.cc index add992d84..c3492f105 100644 --- a/block/testing/content_descriptor_test.cc +++ b/block/testing/content_descriptor_test.cc @@ -147,3 +147,18 @@ BOOST_AUTO_TEST_CASE( test_compressed_random_access ) { } + +BOOST_AUTO_TEST_CASE( test_parallel_decompress_zstd ) +{ + string input_file = "mldb_test_data/Books_5.json.zstd"; + ContentDescriptor descriptor = jsonDecode("file://" + input_file); + std::shared_ptr handler = getDecompressedContent(descriptor); + + auto onBlock = [&] (size_t blockNum, uint64_t blockOffset, + FrozenMemoryRegion block) + { + return true; + }; + + handler->forEachBlockParallel(0, 1024 * 1024 /* requested block size */, 1 /* maxParallelism */, onBlock); +} diff --git a/jml-build/os/Darwin.mk b/jml-build/os/Darwin.mk index fb5293ec9..2b93771a4 100644 --- a/jml-build/os/Darwin.mk +++ b/jml-build/os/Darwin.mk @@ -11,7 +11,7 @@ READLINK:=readlink linker_rpath= SO_EXTENSION:=.dylib -VIRTUALENV ?= virtualenv +VIRTUALENV ?= virtualenv-$(ARCH)-$(OSNAME)-$(PYTHON_VERSION) PYTHON ?= $(VIRTUALENV)/bin/python PIP ?= $(VIRTUALENV)/bin/pip PYTHON_DEPENDENCIES_PRE_CMD ?= $(PIP) install -U pip==21.2.3 diff --git a/jml-build/os/Linux.mk b/jml-build/os/Linux.mk index adb8d13fe..586a67c84 100644 --- a/jml-build/os/Linux.mk +++ b/jml-build/os/Linux.mk @@ -15,7 +15,7 @@ DIST_CODENAME:=$(shell lsb_release -sc) MACHINE_NAME:=$(shell uname -n) READLINK:=readlink -f -VIRTUALENV ?= virtualenv +VIRTUALENV ?= virtualenv-$(ARCH)-$(OSNAME)-$(PYTHON_VERSION) PYTHON ?= $(VIRTUALENV)/bin/python PIP ?= $(VIRTUALENV)/bin/pip PYTHON_DEPENDENCIES_PRE_CMD ?= $(PIP) install -U pip==21.1.3 diff --git a/makefile-main-plugin.mk b/makefile-main-plugin.mk index 25cf765a9..313958c7f 100644 --- a/makefile-main-plugin.mk +++ b/makefile-main-plugin.mk @@ -2,6 +2,7 @@ # Empty default suffixes to speed up initialization .SUFFIXES: +OSNAME:=$(shell uname -s) toolchain ?= gcc PYTHON_ENABLED:=1 @@ -35,27 +36,28 @@ default: all BUILD ?= build ARCH ?= $(shell uname -m) -OBJ := $(BUILD)/$(ARCH)/obj -BIN := $(BUILD)/$(ARCH)/bin +OBJ := $(BUILD)/$(ARCH)-$(OSNAME)/obj +BIN := $(BUILD)/$(ARCH)-$(OSNAME)/bin LIB := $(BUILD)/$(ARCH)/lib -TESTS := $(BUILD)/$(ARCH)/tests -TMPBIN := $(BUILD)/$(ARCH)/tmp -INC := $(BUILD)/$(ARCH)/include +TESTS := $(BUILD)/$(ARCH)-$(OSNAME)/tests +TMPBIN := $(BUILD)/$(ARCH)-$(OSNAME)/tmp +INC := $(BUILD)/$(ARCH)-$(OSNAME)/include SRC := . -TMP ?= $(BUILD)/$(ARCH)/tmp +TMP ?= $(BUILD)/$(ARCH)-$(OSNAME)/tmp # These are for cross-compilation, where binaries used in the build need # be be built for the host. HOSTARCH ?= $(ARCH) -HOSTBIN ?= $(BUILD)/$(HOSTARCH)/bin -HOSTLIB ?= $(BUILD)/$(HOSTARCH)/lib -HOSTINC ?= $(BUILD)/$(HOSTARCH)/include +HOSTOSNAME ?= $(OSNAME) +HOSTBIN ?= $(BUILD)/$(HOSTARCH)-$(HOSTOSNAME)/bin +HOSTLIB ?= $(BUILD)/$(HOSTARCH)-$(HOSTOSNAME)/lib +HOSTINC ?= $(BUILD)/$(HOSTARCH)-$(HOSTOSNAME)/include TEST_TMP := $(TESTS) # Vars for configuration files or files that live outside bin and lib ALTROOT := $(BUILD)/$(ARCH)/altroot ETC := $(ALTROOT)/etc -PLUGINS := $(BUILD)/$(ARCH)/mldb_plugins +PLUGINS := $(BUILD)/$(ARCH)-$(OSNAME)/mldb_plugins JML_BUILD := mldb/jml-build INCLUDE := -I. -Imldb diff --git a/mldb_test_data b/mldb_test_data index 7db85bd10..a5bc078cd 160000 --- a/mldb_test_data +++ b/mldb_test_data @@ -1 +1 @@ -Subproject commit 7db85bd10ecc3acdaa7f37edf08a5e4e56fb3b02 +Subproject commit a5bc078cdb5941498a39b6b3ca0c91be7ce9282e diff --git a/plugins/jml/randomforest.h b/plugins/jml/randomforest.h index 99002970a..a908e3c90 100644 --- a/plugins/jml/randomforest.h +++ b/plugins/jml/randomforest.h @@ -321,11 +321,16 @@ struct PartitionData { bool ordinal = features.at(featureToSplitOn).ordinal; + // Density of example numbers within our set of rows. When this + // gets too low, we do essentially random accesses and it kills + // our cache performance. In that case we can re-index to reduce + // the size. double useRatio = 1.0 * rows.size() / rows.back().exampleNum; //todo: Re-index when usable data fits inside cache - bool reIndex = useRatio < 0.1; + bool reIndex = useRatio < 0.25; //reIndex = false; + //using namespace std; //cerr << "useRatio = " << useRatio << endl; if (!reIndex) { diff --git a/plugins/textual/importtext_procedure.cc b/plugins/textual/importtext_procedure.cc index cfc1edd28..399eb0ca4 100644 --- a/plugins/textual/importtext_procedure.cc +++ b/plugins/textual/importtext_procedure.cc @@ -7,6 +7,7 @@ #include "importtext_procedure.h" #include "mldb/arch/timers.h" +#include "mldb/arch/demangle.h" #include "mldb/utils/csv.h" #include "mldb/utils/lightweight_hash.h" #include "mldb/base/parallel.h" @@ -28,6 +29,7 @@ #include "mldb/base/parse_context.h" #include "mldb/sql/sql_expression_operations.h" #include "mldb/base/optimized_path.h" +#include "mldb/base/hex_dump.h" using namespace std; @@ -552,13 +554,17 @@ struct ImportTextProcedureWorkInstance MldbEngine * engine, const std::function & onProgress) { - string filename = config.dataFileUrl.toDecodedString(); - - // Ask for a memory mappable stream if possible - filter_istream stream(config.dataFileUrl, { { "mapped", "true" } }); - - // Get the file timestamp out - ts = stream.info().lastModified; + // Get a handle to this content, which ensures we don't have any + // kind of version skew from asking for different parts of the file + std::shared_ptr content + = getDecompressedContent(config.dataFileUrl); + + // For some operations, we need a stream. Get this from the + // content handler, so that the underlying data is all shared. + filter_istream stream = content->getStream({ { "mapped", true } }); + + // Get the file timestamp out, to be used internally + ts = content->getLastModified(); std::string line; @@ -649,12 +655,12 @@ struct ImportTextProcedureWorkInstance } try { - ParseContext pcontext(filename, - header.c_str(), header.length(), 1, 0); + ParseContext pcontext(config.dataFileUrl.getUrlStringUtf8(), + header.c_str(), header.length(), 1, 0); fields = expect_csv_row(pcontext, -1, separator); break; } - catch (FileFinishInsideQuote & exp) { + catch (const FileFinishInsideQuote & exp) { if(config.allowMultiLines) { prevHeader.assign(std::move(header)); continue; @@ -666,7 +672,8 @@ struct ImportTextProcedureWorkInstance if (config.autoGenerateHeaders) { // Re-open stream - stream.open(config.dataFileUrl, { { "mapped", "true" } }); + content = getContent(config.dataFileUrl); + stream = content->getStream({ { "mapped", true } }); auto nfields = fields.size(); for (ssize_t i = 0; i < nfields; ++i) { inputColumnNames.emplace_back(i); @@ -714,7 +721,7 @@ struct ImportTextProcedureWorkInstance // Now we know the columns, we can bind our SQL expressions for the // select, where, named and timestamp parts of the expression. SqlCsvScope scope(engine, inputColumnNames, ts, - Utf8String(config.dataFileUrl.toDecodedString())); + config.dataFileUrl.getUrlString()); selectBound = config.select.bind(scope); whereBound = config.where->bind(scope); @@ -783,17 +790,33 @@ struct ImportTextProcedureWorkInstance << jsonEncodeStr(knownColumnNames); // Skip those up to the offset now we've done the header + // TODO: do this skipping later on for (size_t i = 0; stream && i < config.offset; ++i, ++lineOffset) { getline(stream, line); } - loadTextData(dataset, stream, config, scope, onProgress); + if (stream.eof()) { + // Empty lines? EOF + return; + } + + auto offset = stream.tellg(); + if (offset == -1) { + cerr << type_name(*stream.rdbuf()) << endl; + throw AnnotatedException + (400, "Stream for import text must be able to tell its offset", + "streambuf", type_name(*stream.rdbuf())); + } + + loadTextData(dataset, content, offset, + config, scope, onProgress); } /* Load, filter and format all lines and process them */ void loadTextData(std::shared_ptr dataset, - std::istream& stream, + std::shared_ptr content, + uint64_t offset, const ImportTextConfig& config, SqlCsvScope& scope, const std::function & onProgress) @@ -978,6 +1001,11 @@ struct ImportTextProcedureWorkInstance return true; } + cerr << "lineNumber " << lineNumber << endl; + cerr << "columnNumber " << columnNumber << endl; + cerr << "line " << line << endl; + hex_dump(line.data(), line.length()); + throw AnnotatedException(400, "Error parsing CSV row: " + message, "lineNumber", lineNumber, @@ -1007,13 +1035,20 @@ struct ImportTextProcedureWorkInstance /// Bytes done in this thread uint64_t bytesDone = 0; + + /// Number of lines in chunk + ssize_t numLinesInChunk = -1; }; PerThreadAccumulator accum; - auto startChunk = [&] (int64_t chunkNumber, size_t lineNumber) + std::atomic deferredEmptyLineNumber(-1); + + auto startChunk = [&] (int64_t chunkNumber, size_t lineNumber, + ssize_t numLines) { auto & threadAccum = accum.get(); + threadAccum.numLinesInChunk = numLines; threadAccum.threadRecorder = recorder.newChunk(chunkNumber); if (isIdentitySelect || canUseDecomposed) threadAccum.specializedRecorder @@ -1080,11 +1115,19 @@ struct ImportTextProcedureWorkInstance config.skipLineRegex)) return true; } - + // MLDB-1111 empty lines are treated as error - if (length == 0) - return handleError("empty line", actualLineNum, 0, ""); - + if (length == 0) { + if (lineNum == threadAccum.numLinesInChunk - 1 + && deferredEmptyLineNumber.exchange(actualLineNum) == -1) { + // may not be an empty line error, since last line in + // chunk + return true; // don't record this row + } + else { + return handleError("empty line", actualLineNum, 0, ""); + } + } // Values that come in from the CSV file PossiblyDynamicBuffer values(inputColumnNames.size()); @@ -1104,21 +1147,21 @@ struct ImportTextProcedureWorkInstance config.processExcelFormulas, scope.columnsUsed); - if (errorMsg) { - if(config.allowMultiLines) { - // check if we hit an error meaning we probably - // have a multiline error - if(errorMsg == unclosedQuoteError || - errorMsg == notEnoughColsError) { - return false; - } + if (errorMsg) { + if(config.allowMultiLines) { + // check if we hit an error meaning we probably + // have a multiline error + if(errorMsg == unclosedQuoteError || + errorMsg == notEnoughColsError) { + return false; } - - return handleError(errorMsg, actualLineNum, - line - lineStart + 1, - string(line, length)); } + return handleError(errorMsg, actualLineNum, + line - lineStart + 1, + string(line, length)); + } + auto row = scope.bindRow(values.data(), ts, actualLineNum, 0 /* todo: chunk ofs */); @@ -1311,22 +1354,41 @@ struct ImportTextProcedureWorkInstance if(!config.allowMultiLines) { - forEachLineBlock(stream, onLine, config.limit, + forEachLineBlock(content, offset, onLine, config.limit, numCpus() /* parallelism */, startChunk, doneChunk); } else { + + auto stream = content->getStream(); + + { + //stream.seekg(offset, std::ios_base::cur); + // not all streams support seeking + + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + size_t currentOffset = 0; + + while (currentOffset < offset) { + size_t n = std::min(offset - currentOffset, BUFFER_SIZE); + stream.read(buffer, n); + currentOffset += stream.gcount(); + } + } + // very simplistic and not efficient way of doing multi-line. we send // lines one by one to the 'onLine' function, and if // we get an error that probably is caused by a multi- // line string, we concat the current line with the next // one and try again. - startChunk(0, 0); + startChunk(0, 0, -1 /* num lines is unknown */); string line; string t_line; string prevLine; int64_t lineNum = 0; + while(getline(stream, line)) { // prepend previous line if we're tagging it along if(!prevLine.empty()) { @@ -1350,6 +1412,11 @@ struct ImportTextProcedureWorkInstance doneChunk(0, lineNum); } + if (deferredEmptyLineNumber != -1 + && deferredEmptyLineNumber < lineCount - 1) { + handleError("empty line", deferredEmptyLineNumber, 0, ""); + } + // Accumulate any from the end accum.forEach([&] (ThreadAccum * accum) { diff --git a/plugins/textual/importtext_procedure.h b/plugins/textual/importtext_procedure.h index 13954ca8e..a92b05ec0 100644 --- a/plugins/textual/importtext_procedure.h +++ b/plugins/textual/importtext_procedure.h @@ -15,14 +15,18 @@ #include "mldb/core/function.h" #include "mldb/types/optional.h" #include "mldb/types/regex.h" +#include "mldb/block/content_descriptor.h" namespace MLDB { +/*****************************************************************************/ +/* IMPORT TEXT CONFIG */ +/*****************************************************************************/ + struct ImportTextConfig : public ProcedureConfig { static constexpr const char * name = "import.text"; - - Url dataFileUrl; + ContentDescriptor dataFileUrl; PolyConfigT outputDataset = DefaultType("tabular"); std::vector headers; std::string delimiter = ","; diff --git a/plugins/textual/textual.mk b/plugins/textual/textual.mk index 661aa6556..79a8ac60b 100644 --- a/plugins/textual/textual.mk +++ b/plugins/textual/textual.mk @@ -13,7 +13,6 @@ LIBMLDB_TEXTUAL_PLUGIN_SOURCES:= \ sql_csv_scope.cc \ tokensplit.cc \ - LIBMLDB_TEXTUAL_PLUGIN_LINK:= \ mldb_core \ mldb_engine \ @@ -36,6 +35,7 @@ LIBMLDB_TEXTUAL_PLUGIN_LINK:= \ mldb_builtin_base \ mldb_builtin \ sql_types \ + block \ $(eval $(call library,mldb_textual_plugin,$(LIBMLDB_TEXTUAL_PLUGIN_SOURCES),$(LIBMLDB_TEXTUAL_PLUGIN_LINK))) diff --git a/testing/MLDB-1395-error-message-file-doesnt-exist.js b/testing/MLDB-1395-error-message-file-doesnt-exist.js index 694e4da5b..1814067ff 100644 --- a/testing/MLDB-1395-error-message-file-doesnt-exist.js +++ b/testing/MLDB-1395-error-message-file-doesnt-exist.js @@ -18,7 +18,8 @@ var resp = mldb.put("/v1/procedures/csv_proc", config) mldb.log(resp); -unittest.assertEqual(resp.json.details.runError.error.indexOf("Opening file ./thisfiledoesnotexist: failed opening file: No such file or directory"), 0); +unittest.assertEqual(resp.json.details.runError.error.indexOf("No such file or directory") >= 0, true); +unittest.assertEqual(resp.json.details.runError.error.indexOf("thisfiledoesnotexist") >= 0, true); "success" diff --git a/testing/MLDB-749-csv-dataset.js b/testing/MLDB-749-csv-dataset.js index 24af39a31..357d6102f 100644 --- a/testing/MLDB-749-csv-dataset.js +++ b/testing/MLDB-749-csv-dataset.js @@ -44,7 +44,15 @@ mldb.log(res.json); csv_conf = { type: "import.text", params: { - dataFileUrl : "https://raw.githubusercontent.com/datacratic/mldb-pytanic-plugin/master/titanic_train.csv", + dataFileUrl : { + url: "https://raw.githubusercontent.com/datacratic/mldb-pytanic-plugin/master/titanic_train.csv", + etag: { + authority: "https://raw.githubusercontent.com", + value: "fdb7d4717c5befa93d0f241ac4245bed1a2a10e7" + }, + sha256: "31ad156ab55993d901bd607828045a3de18cac4144be6dc1c520bc41573a8115", + sha3: "9a87daef2da4915211114d4cfecae175636a8074ee6dbd17483c237f" + }, outputDataset: { id: "titanic", }, @@ -52,7 +60,11 @@ csv_conf = { } } -var res = mldb.put("/v1/procedures/csv_proc", csv_conf) +var res = mldb.put("/v1/procedures/csv_proc", csv_conf); + +mldb.log(res); + +unittest.assertEqual(res.responseCode, 201); var res = mldb.get('/v1/datasets/titanic/query', { limit: 10, format: 'table', orderBy: 'rowName()'}); diff --git a/testing/dataset/iris.data b/testing/dataset/iris.data index 5c4316cd6..a3490e0e0 100755 --- a/testing/dataset/iris.data +++ b/testing/dataset/iris.data @@ -148,4 +148,3 @@ 6.5,3.0,5.2,2.0,Iris-virginica 6.2,3.4,5.4,2.3,Iris-virginica 5.9,3.0,5.1,1.8,Iris-virginica - diff --git a/utils/for_each_line.cc b/utils/for_each_line.cc index e16f0897c..c6e3e8389 100644 --- a/utils/for_each_line.cc +++ b/utils/for_each_line.cc @@ -486,7 +486,6 @@ void forEachLineBlock(std::istream & stream, } } - /*****************************************************************************/ /* FOR EACH LINE BLOCK (CONTENT HANDLER) */ /*****************************************************************************/ @@ -602,7 +601,7 @@ void forEachLineBlock(std::shared_ptr content, size_t length = mem.length(); const char * start = mem.data(); - const char * current = length == 0 ? start : (const char *)memchr(start, '\n', length); + const char * current = (const char *)memchr(start, '\n', length); const char * end = start + length; size_t numLinesInBlock = 0; @@ -843,7 +842,6 @@ void forEachLineBlock(std::shared_ptr content, } } #endif - } // If there was an exception, rethrow it rather than returning @@ -854,7 +852,6 @@ void forEachLineBlock(std::shared_ptr content, } - /*****************************************************************************/ /* FOR EACH CHUNK */ /*****************************************************************************/ diff --git a/vfs/compressor.cc b/vfs/compressor.cc index 8ec35f30e..6c505d6d9 100644 --- a/vfs/compressor.cc +++ b/vfs/compressor.cc @@ -13,6 +13,7 @@ #include #include #include "mldb/base/thread_pool.h" +#include using namespace std; @@ -44,6 +45,20 @@ std::map decompressors; } // file scope +bool +Compressor:: +canFixupLength() const +{ + return false; +} + +std::string +Compressor:: +newHeaderForLength(uint64_t lengthWritten) const +{ + throw MLDB::Exception("Attempt to call newHeaderForLength for class that doesn't support it"); +} + std::string Compressor:: filenameToCompression(const std::string & filename) @@ -175,7 +190,23 @@ forEachBlockParallel(size_t requestedBlockSize, std::shared_ptr buf; ThreadWorkGroup tp(maxParallelism); - + + // We queue them up as we want to ensure they run in sequence + std::mutex jobsMutex; + std::deque jobs; + + auto doOne = [&] () + { + ThreadJob job; + { + std::unique_lock guard{jobsMutex}; + ExcAssert(!jobs.empty()); + job = std::move(jobs.front()); + jobs.pop_front(); + } + job(); + }; + while (std::get<0>((std::tie(buf, numChars) = getData(requestedBlockSize)))) { auto onData = [&] (std::shared_ptr data, size_t len) -> size_t { @@ -196,7 +227,11 @@ forEachBlockParallel(size_t requestedBlockSize, }; if (maxParallelism > 0) { - tp.add(std::move(doBlock)); + { + std::unique_lock guard{jobsMutex}; + jobs.push_back(std::move(doBlock)); + } + tp.add(doOne); } else { doBlock(); diff --git a/vfs/compressor.h b/vfs/compressor.h index c2c3a3f64..7eb2ce9f9 100644 --- a/vfs/compressor.h +++ b/vfs/compressor.h @@ -44,7 +44,7 @@ struct Compressor { passed to compress. Some of the compression formats can put this into their header to enable better decisions on allocation. - This must be called either never or ponce before compress(), flush() or + This must be called either never or once before compress(), flush() or finish(), and behavior is undefined if the exact same amount of data is not then passed to compress(). @@ -69,6 +69,19 @@ struct Compressor { */ virtual void finish(const OnData & onData) = 0; + /** Returns true if the compressor is able to write a new header with + a fixed-up length once closed. + + Default implementation returns false. + */ + virtual bool canFixupLength() const; + + /** Return a new header that will overwrite the one written before. The + * new header MUST have the same length as the old one. Will only be + * called if canFixupLength() returns true. + */ + virtual std::string newHeaderForLength(uint64_t lengthWritten) const; + /** Convert a filename to a compression scheme. Returns the empty string if it isn't found. */ diff --git a/vfs/filter_streams.cc b/vfs/filter_streams.cc index a8138608c..e03e07092 100644 --- a/vfs/filter_streams.cc +++ b/vfs/filter_streams.cc @@ -75,9 +75,19 @@ UriHandler(std::streambuf * buf, struct BoostCompressor: public boost::iostreams::multichar_output_filter { - BoostCompressor(Compressor * compressor) - : compressor(compressor) + BoostCompressor(Compressor * compressor, std::streambuf & buf) + : compressor(compressor), buf(buf) { + // Record the start position so that once we know the actual written + // size we can go back and rewrite the header wiht the right size. + if (compressor->canFixupLength()) { + try { + MLDB_TRACE_EXCEPTIONS(false); + startPos = buf.pubseekoff(0, ios::cur, ios_base::in); + } catch (std::ios_base::failure exc) { + startPos = -1; + } + } } template @@ -89,6 +99,7 @@ struct BoostCompressor: public boost::iostreams::multichar_output_filter { data += written; size -= written; + bytesWritten += written; } } @@ -102,19 +113,40 @@ struct BoostCompressor: public boost::iostreams::multichar_output_filter { }; compressor->compress(s, n, onData); + bytesRead += n; return n; } template void close(Sink& sink) { + if (alreadyClosed) + return; + auto onData = [&] (const char * data, size_t len) -> size_t { writeAll(sink, data, len); return len; }; + if (bytesRead == 0) { + compressor->notifyInputSize(0); + } compressor->finish(onData); + + if (bytesRead != 0 && startPos != -1 && compressor->canFixupLength()) { + std::string data = compressor->newHeaderForLength(bytesRead); + { + std::ostream stream(&buf); + auto oldPos = stream.tellp(); + stream.seekp(startPos, std::ios::beg); + stream.write(data.data(), data.size()); + stream.flush(); + stream.seekp(oldPos, std::ios::beg); + } + } + + alreadyClosed = true; } template @@ -130,6 +162,11 @@ struct BoostCompressor: public boost::iostreams::multichar_output_filter { } std::shared_ptr compressor; + std::streambuf & buf; + uint64_t bytesWritten = 0; + uint64_t bytesRead = 0; + int64_t startPos = -1; + bool alreadyClosed = false; }; @@ -361,7 +398,7 @@ void addCompression(streambuf & buf, = Compressor::create(compression, compressionLevel); if (!compressor) throw MLDB::Exception("unknown filter compression " + compression); - stream.push(BoostCompressor(compressor)); + stream.push(BoostCompressor(compressor, buf)); } else { std::string compressionFromFilename @@ -371,7 +408,7 @@ void addCompression(streambuf & buf, = Compressor::create(compressionFromFilename, compressionLevel); if (!compressor) throw MLDB::Exception("unknown filter compression " + compression); - stream.push(BoostCompressor(compressor)); + stream.push(BoostCompressor(compressor, buf)); } } } @@ -635,11 +672,11 @@ close() { if (stream) { boost::iostreams::flush(*stream); - boost::iostreams::close(*stream); + //boost::iostreams::close(*stream); // will be called on stream.reset(); } + stream.reset(); exceptions(ios::goodbit); rdbuf(0); - stream.reset(); sink.reset(); options.clear(); if (deferredExcPtr) { diff --git a/vfs/libdb_intialization.cc b/vfs/libdb_intialization.cc new file mode 100644 index 000000000..381929a26 --- /dev/null +++ b/vfs/libdb_intialization.cc @@ -0,0 +1,52 @@ +/* libdb_initialization.cc + Jeremy Barnes, 13 March 2005 + Copyright (c) 2005 Jeremy Barnes. All rights reserved. + This file is part of MLDB. Copyright 2015 mldb.ai inc. All rights reserved. + + Initialization of stream handlers in the presence of filter_xstream. +*/ + +#include "mldb/types/db/portable_iarchive.h" +#include "mldb/types/db/portable_oarchive.h" +#include "filter_streams.h" + +namespace MLDB { + +namespace { + +// Plug in more capable filter_stream classes instead of the ofstream that +// libdb comes with. +struct AtInit { + AtInit() + { + oldOpenInputStream = defaultOpenInputStream; + oldOpenOutputStream = defaultOpenOutputStream; + + defaultOpenInputStream = open_input; + defaultOpenOutputStream = open_output; + } + + ~AtInit() + { + defaultOpenInputStream = oldOpenInputStream; + defaultOpenOutputStream = oldOpenOutputStream; + } + + static std::istream * open_input(const std::string & filename) + { + return new filter_istream(filename); + } + + static std::ostream * open_output(const std::string & filename) + { + return new filter_ostream(filename); + } + + std::function oldOpenInputStream; + std::function oldOpenOutputStream; + +} atInit; + +} // file scope + +} // namespace MLDB diff --git a/vfs/lz4.cc b/vfs/lz4.cc index a08db19df..8b1a9185f 100644 --- a/vfs/lz4.cc +++ b/vfs/lz4.cc @@ -159,6 +159,19 @@ struct MLDB_PACKED Header return sizeof(*this); } + std::string asString(uint64_t lengthWritten) + { + std::string result; + const std::function onData + = [&] (const char * data, size_t len) -> size_t + { + result.append(data, len); + return len; + }; + write(onData, lengthWritten); + return result; + } + uint8_t checksumOptions(const uint64_le & knownContentSize) const { if (contentSize()) { @@ -192,7 +205,7 @@ struct Lz4Compressor : public Compressor { typedef Compressor::FlushLevel FlushLevel; Lz4Compressor(int level, uint8_t blockSizeId = 7, - uint64_t contentSize = 0) + uint64_t contentSize = -1) : head(blockSizeId, true /* independent blocks */, false /* block checksum */, @@ -222,16 +235,30 @@ struct Lz4Compressor : public Compressor { virtual void notifyInputSize(uint64_t inputSize) override { if (!writeHeader) { - throw Exception("lz4 input size already notified"); + throw Exception("lz4 input size notified too late"); } head.setContentSize(true); this->contentSize = inputSize; } - + + virtual bool canFixupLength() const override + { + return true; + } + + virtual std::string newHeaderForLength(uint64_t lengthWritten) const override + { + lz4::Header fixedHeader = this->head; + fixedHeader.setContentSize(lengthWritten); + std::string result = fixedHeader.asString(lengthWritten); + return result; + } + virtual void compress(const char * s, size_t n, const OnData & onData) override { if (writeHeader) { + auto asStringDebug = head.asString(contentSize); head.write(onData, contentSize); writeHeader = false; } diff --git a/vfs/testing/compressor_test.cc b/vfs/testing/compressor_test.cc index 32640912b..035e9a197 100644 --- a/vfs/testing/compressor_test.cc +++ b/vfs/testing/compressor_test.cc @@ -89,3 +89,28 @@ BOOST_AUTO_TEST_CASE( test_compress_decompress_lz4_content_size ) BOOST_CHECK_EQUAL(size, stream2.readAll().length()); } + +BOOST_AUTO_TEST_CASE( test_compress_decompress_zstd_content_size ) +{ + string input_file = "mldb/vfs/testing/filter_streams_test.cc"; + string output_file = "tmp/compressor_test.zstd"; + string zstd_cmd = binDir / string("zstd -f " + input_file + " -o " + output_file); + + Scope_Exit(::unlink(output_file.c_str())); + system(zstd_cmd); + + std::ifstream stream(output_file.c_str()); + + char buf[200]; + stream.read(buf, 200); + + int numRead = stream.gcount(); + + auto decomp = Decompressor::create("zstd"); + + auto size = decomp->decompressedSize(buf, numRead, -1 /* total len unknown */); + + filter_istream stream2(output_file); + + BOOST_CHECK_EQUAL(size, stream2.readAll().length()); +} diff --git a/vfs/testing/filter_streams_test.cc b/vfs/testing/filter_streams_test.cc index ff3f46f22..1c9dd7c98 100644 --- a/vfs/testing/filter_streams_test.cc +++ b/vfs/testing/filter_streams_test.cc @@ -29,7 +29,9 @@ #include "mldb/base/scope.h" #include "mldb/arch/exception_handler.h" +#include "mldb/base/exc_assert.h" #include "mldb/arch/demangle.h" +#include "mldb/base/hex_dump.h" using namespace std; namespace fs = std::filesystem; @@ -37,6 +39,14 @@ using namespace MLDB; using boost::unit_test::test_suite; +struct OnInit { + OnInit() + { + ExcAssert(getenv("BIN")); + ExcAssert(getenv("TMP")); + } +} onInit; + fs::path binDir = std::string(getenv("BIN")); fs::path tmpDir = std::string(getenv("TMP")); @@ -69,14 +79,24 @@ void compress_using_tool(const std::string & input_file, const std::string & output_file, const std::string & command) { - system("cat " + input_file + " | " + command + " > " + output_file); + system(command + " " + input_file + " > " + output_file); } void decompress_using_tool(const std::string & input_file, const std::string & output_file, const std::string & command) { - system("cat " + input_file + " | " + command + " > " + output_file); + try { + //system("hexdump -C " + input_file + " | head -n 10"); + system("cat " + input_file + " | " + command + " > " + output_file); + } catch (...) { + std::ifstream stream(input_file); + constexpr size_t BUF_SIZE = 1024; + char buf[BUF_SIZE]; + size_t n = stream.readsome(buf, BUF_SIZE); + hex_dump(buf, n); + throw; + } } void compress_using_stream(const std::string & input_file, @@ -86,7 +106,7 @@ void compress_using_stream(const std::string & input_file, filter_ostream out(output_file); - char buf[16386]; + char buf[16384]; while (in) { in.read(buf, 16384); @@ -164,19 +184,19 @@ void test_compress_decompress(const std::string & input_file, BOOST_AUTO_TEST_CASE( test_compress_decompress_gz ) { string input_file = "mldb/vfs/testing/filter_streams_test.cc"; - test_compress_decompress(input_file, "gz", "gzip", "gzip -d"); + test_compress_decompress(input_file, "gz", "gzip -c", "gzip -d"); } BOOST_AUTO_TEST_CASE( test_compress_decompress_bzip2 ) { string input_file = "mldb/vfs/testing/filter_streams_test.cc"; - test_compress_decompress(input_file, "bz2", "bzip2", "bzip2 -d"); + test_compress_decompress(input_file, "bz2", "bzip2 -c", "bzip2 -d"); } BOOST_AUTO_TEST_CASE( test_compress_decompress_xz ) { string input_file = "mldb/vfs/testing/filter_streams_test.cc"; - test_compress_decompress(input_file, "xz", "xz", "xz -d"); + test_compress_decompress(input_file, "xz", "xz -c", "xz -d"); } BOOST_AUTO_TEST_CASE( test_compress_decompress_lz4 ) @@ -189,15 +209,15 @@ BOOST_AUTO_TEST_CASE( test_compress_decompress_lz4 ) BOOST_AUTO_TEST_CASE( test_compress_decompress_lz4_content_size ) { string input_file = "mldb/vfs/testing/filter_streams_test.cc"; - string lz4_cmd = binDir / "lz4cli --content-size"; - test_compress_decompress(input_file, "lz4", lz4_cmd, lz4_cmd + " -d"); + string lz4_cmd = binDir / "lz4cli -B7 -BX --content-size"; + test_compress_decompress(input_file, "csize.lz4", lz4_cmd, lz4_cmd + " -d"); } BOOST_AUTO_TEST_CASE( test_compress_decompress_zstandard ) { string input_file = "mldb/vfs/testing/filter_streams_test.cc"; string zstd_cmd = binDir / "zstd"; - test_compress_decompress(input_file, "zst", zstd_cmd, zstd_cmd + " -d"); + test_compress_decompress(input_file, "zst", zstd_cmd + " -c", zstd_cmd + " -d"); } BOOST_AUTO_TEST_CASE( test_open_failure ) diff --git a/vfs/testing/vfs_testing.mk b/vfs/testing/vfs_testing.mk index f0cea4471..3bf174973 100644 --- a/vfs/testing/vfs_testing.mk +++ b/vfs/testing/vfs_testing.mk @@ -1,7 +1,7 @@ # This file is part of MLDB. Copyright 2015 mldb.ai inc. All rights reserved. -$(eval $(call test,filter_streams_test,vfs $(STD_FILESYSTEM_LIBNAME) boost_system value_description arch types,boost)) +$(eval $(call test,filter_streams_test,vfs $(STD_FILESYSTEM_LIBNAME) boost_system value_description arch base types,boost)) $(TESTS)/filter_streams_test: $(BIN)/lz4cli $(BIN)/zstd -$(eval $(call test,compressor_test,vfs $(STD_FILESYSTEM_LIBNAME) boost_system value_description arch types,boost)) +$(eval $(call test,compressor_test,vfs $(STD_FILESYSTEM_LIBNAME) boost_system value_description arch base types,boost)) diff --git a/vfs/zstandard.cc b/vfs/zstandard.cc index 03531704d..d28b68683 100644 --- a/vfs/zstandard.cc +++ b/vfs/zstandard.cc @@ -6,11 +6,17 @@ Zstandard compressor and decompressors. */ +#define ZSTD_STATIC_LINKING_ONLY 1 + #include "compressor.h" #include "mldb/base/exc_assert.h" #include "mldb/ext/zstd/lib/zstd.h" +#include "mldb/base/thread_pool.h" +#include "mldb/arch/exception.h" +#include "mldb/arch/endian.h" #include #include +#include using namespace std; @@ -130,15 +136,14 @@ struct ZStandardDecompressor: public Decompressor { virtual int64_t decompressedSize(const char * block, size_t blockLen, int64_t totalLen) const override { - unsigned long long res = ZSTD_getDecompressedSize(block, blockLen); - if (ZSTD_isError(res)) { - return LENGTH_INSUFFICIENT_DATA; - } - else if (res == 0) { + unsigned long long res = ZSTD_getFrameContentSize(block, blockLen); + if (res == ZSTD_CONTENTSIZE_UNKNOWN) { return LENGTH_UNKNOWN; // 0 means zero length OR unknown } - else return res; - + else if (res == ZSTD_CONTENTSIZE_ERROR) { + return LENGTH_INSUFFICIENT_DATA; + } + else return res; } virtual void decompress(const char * data, size_t len, @@ -159,7 +164,135 @@ struct ZStandardDecompressor: public Decompressor { } } } - + + virtual bool + forEachBlockParallel(size_t requestedBlockSize, + const GetDataFunction & getData, + const ForEachBlockFunction & onBlock, + const Allocate & allocate, + int maxParallelism) override + { + return Decompressor::forEachBlockParallel(requestedBlockSize, getData, onBlock, allocate, maxParallelism); + // To do this, we would need to: + // - Get fixed sized chunks from the underlying stream + // - Pass each to a stream + // - For each one, scan for the magic characters to indicate the beginning + // - Decompress the blocks (taking into account the possibility that there are + // spurious magic markers) + // - Scan into the next one's blocks to finish off the last one + // - Verify that each before and after block agrees on what they are scanning for +#if 0 + ThreadWorkGroup tp(maxParallelism); + + std::shared_ptr lastBlock; + std::shared_ptr currentBlock; + const char * currentData = nullptr; + size_t currentBlockLeft = 0; + + auto makeAvail = [&] (size_t numBytes) -> std::span + { + if (numBytes > currentBlockLeft) { + auto [newBlock, newBlockSize] = getData(numBytes); + if (currentBlockLeft == 0) { + // First block or previous one completely used + cerr << "first block" << endl; + currentBlock = std::move(newBlock); + currentBlockLeft = newBlockSize; + currentData = currentBlock.get(); + } + else { + if (newBlock.get() == currentData + currentBlockLeft) { + // Contiguous; must be mapped + cerr << "contiguous" << endl; + lastBlock = std::move(currentBlock); + currentBlock = std::move(newBlock); + currentBlockLeft += newBlockSize; + } + else { + // Combine them together (slow, should try to avoid making this happen) + cerr << "combined" << endl; + std::shared_ptr combinedBlock + (new char[newBlockSize + currentBlockLeft], + [] (auto p) { delete[] p; }); + memcpy(combinedBlock.get(), currentData, currentBlockLeft); + memcpy(combinedBlock.get() + currentBlockLeft, newBlock.get(), newBlockSize); + lastBlock = {}; + currentBlock = std::move(combinedBlock); + currentData = combinedBlock.get(); + currentBlockLeft += newBlockSize; + } + } + } + + ExcAssertLessEqual(numBytes, currentBlockLeft); + + return { currentData, numBytes }; + }; + + auto consume = [&] (size_t numBytes) + { + ExcAssertGreaterEqual(currentBlockLeft, numBytes); + currentData += numBytes; + currentBlockLeft -= numBytes; + }; + + //auto magicBytes = makeAvail(4); + //uint32_le magic; + //std::memcpy(&magic, magicBytes.data(), 4); + //if (magic != ZSTD_MAGICNUMBER) + // throw MLDB::Exception("Corrupted zstandard file: magic number is wrong"); + + std::shared_ptr context(ZSTD_createDCtx(), ZSTD_freeDCtx); + std::span header = makeAvail(ZSTD_FRAMEHEADERSIZE_MAX); + ZSTD_frameHeader frameHeader; + cerr << "Frame header: max size " << ZSTD_FRAMEHEADERSIZE_MAX << endl; + hex_dump(header.data(), header.size()); + size_t headerResult = ZSTD_getFrameHeader(&frameHeader, header.data(), header.size()); + if (headerResult != 0) + throw MLDB::Exception("Corrupted ZStandard file: frame header not available"); + //consume(frameHeader.headerSize); + + cerr << "dict id " << frameHeader.dictID << endl; + cerr << "content size " << frameHeader.frameContentSize << endl; + cerr << "window size " << frameHeader.windowSize << endl; + cerr << "frame type " << frameHeader.frameType << endl; + cerr << "header size " << frameHeader.headerSize << endl; + cerr << "block size max " << frameHeader.blockSizeMax << endl; + + size_t outputBufSize = frameHeader.windowSize * 2; + std::shared_ptr outputBuf(new char[outputBufSize]); + char * outputBufPtr = outputBuf.get(); + char * outputBufEnd = outputBufPtr + outputBufSize; + + size_t res = ZSTD_decompressBegin(context.get()); + //cerr << "res = " << res << endl; + if (ZSTD_isError(res)) { + throw Exception("Error beginning decompression: %s", + ZSTD_getErrorName(res)); + } + for (size_t i = 0; i < 5; ++i) { + size_t nextSize = ZSTD_nextSrcSizeToDecompress(context.get()); + cerr << "next size is " << nextSize << endl; + if (ZSTD_isError(nextSize)) { + throw Exception("Error getting size to decompress: %s", + ZSTD_getErrorName(nextSize)); + } + auto data = makeAvail(nextSize); + consume(nextSize); + size_t written = ZSTD_decompressContinue(context.get(), outputBufPtr, outputBufEnd - outputBufPtr, + data.data(), data.size()); + cerr << "written " << written << " bytes" << endl; + if (ZSTD_isError(written)) { + throw Exception("Error continuing decompression: %s", + ZSTD_getErrorName(written)); + } + outputBufPtr += written; + } + + throw MLDB::Exception("not finished"); +#endif + } + virtual void finish(const OnData & onData) override { }