Removed protobufs.

Author: jgates108

src/CMakeLists.txt

@@ -66,7 +66,6 @@ add_subdirectory(mimic)
 add_subdirectory(mysql)
 add_subdirectory(parser)
 add_subdirectory(partition)
-add_subdirectory(proto)
 add_subdirectory(protojson)
 add_subdirectory(proxy)
 add_subdirectory(qana)

src/ccontrol/CMakeLists.txt

@@ -1,5 +1,4 @@
 add_library(ccontrol SHARED)
-add_dependencies(ccontrol proto)
 
 target_include_directories(ccontrol PRIVATE
     ${ANTLR4_INCLUDE_DIR}
@@ -29,7 +28,6 @@ target_link_libraries(ccontrol PUBLIC
     css
     log
     parser
-	proto
     sphgeom
 )
 

src/ccontrol/MergingHandler.cc

@@ -46,8 +46,6 @@
 #include "http/ClientConnPool.h"
 #include "http/Method.h"
 #include "mysql/CsvBuffer.h"
-#include "proto/ProtoHeaderWrap.h"
-#include "proto/worker.pb.h"
 #include "qdisp/CzarStats.h"
 #include "qdisp/Executive.h"
 #include "qdisp/JobQuery.h"
@@ -145,106 +143,6 @@ string readHttpFileAndMerge(lsst::qserv::qdisp::UberJob::Ptr const& uberJob, str
                               ", offset: " + to_string(offset) + ", ex: " + string(ex.what());
         LOGS(_log, LOG_LVL_ERROR, context << errMsg);
         return errMsg;
-        /* &&&
-                // A value of the flag is set by the message processor when it's time to finish
-                // or abort reading  the file.
-                bool last = false;
-                char const* next = inBuf;
-                char const* const end = inBuf + inBufSize;
-                LOGS(_log, LOG_LVL_TRACE, context << " next=" << (uint64_t)next << " end=" << (uint64_t)end);
-                while ((next < end) && !last) {
-                    if (exec->getCancelled()) {
-                        throw runtime_error(context + " query was cancelled");
-                    }
-                    if (msgSizeBytes == 0) {
-                        // Continue or finish reading the frame header.
-                        size_t const bytes2read =
-                                std::min(sizeof(uint32_t) - msgSizeBufNext, (size_t)(end - next));
-                        std::memcpy(msgSizeBuf.data() + msgSizeBufNext, next, bytes2read);
-                        next += bytes2read;
-                        offset += bytes2read;
-                        msgSizeBufNext += bytes2read;
-                        if (msgSizeBufNext == sizeof(uint32_t)) {
-                            ++headerCount;
-                            // Done reading the frame header.
-                            msgSizeBufNext = 0;
-                            // Parse and evaluate the message length.
-                            msgSizeBytes = *(reinterpret_cast<uint32_t*>(msgSizeBuf.data()));
-                            if (msgSizeBytes == 0) {
-                                throw runtime_error("message size is 0 at offset " +
-                                                    to_string(offset - sizeof(uint32_t)) + ", file: " +
-        httpUrl);
-                            }
-                            if (msgSizeBytes > ProtoHeaderWrap::PROTOBUFFER_HARD_LIMIT) {
-                                throw runtime_error("message size " + to_string(msgSizeBytes) + " at offset "
-        + to_string(offset - sizeof(uint32_t)) + " exceeds the hard limit of " +
-                                                    to_string(ProtoHeaderWrap::PROTOBUFFER_HARD_LIMIT) +
-                                                    ", file: " + httpUrl);
-                            }
-                            // Extend the message buffer (if needed). Note that buffer never gets
-                            // truncated to avoid excessive memory deallocations/allocations.
-                            if (msgBufSize < msgSizeBytes) {
-                                msgBufSize = msgSizeBytes;
-                                msgBuf.reset(new char[msgBufSize]);
-                            }
-                            // Starts the tracker to measure the performance of the network I/O.
-                            transmitRateTracker =
-                                    make_unique<lsst::qserv::TimeCountTracker<double>>(reportFileRecvRate);
-                        }
-                    } else {
-                        // Continue or finish reading the message body.
-                        size_t const bytes2read =
-                                std::min((size_t)msgSizeBytes - msgBufNext, (size_t)(end - next));
-                        std::memcpy(msgBuf.get() + msgBufNext, next, bytes2read);
-                        next += bytes2read;
-                        offset += bytes2read;
-                        msgBufNext += bytes2read;
-                        if (msgBufNext == msgSizeBytes) {
-                            // Done reading message body.
-                            msgBufNext = 0;
-
-                            // Destroying the tracker will result in stopping the tracker's timer and
-                            // reporting the file read rate before proceeding to the merge.
-                            if (transmitRateTracker != nullptr) {
-                                transmitRateTracker->addToValue(msgSizeBytes);
-                                transmitRateTracker->setSuccess();
-                                transmitRateTracker.reset();
-                            }
-
-                            // Parse and evaluate the message.
-                            mergeHappened = true;
-                            bool messageReadyResult = messageIsReady(msgBuf.get(), msgSizeBytes, last);
-                            totalBytesRead += msgSizeBytes;
-                            if (!messageReadyResult) {
-                                success = false;
-                                throw runtime_error("message processing failed at offset " +
-                                                    to_string(offset - msgSizeBytes) + ", file: " + httpUrl);
-                            }
-                            // Reset the variable to prepare for reading the next header & message (if any).
-                            msgSizeBytes = 0;
-                        } else {
-                            LOGS(_log, LOG_LVL_TRACE,
-                                 context << " headerCount=" << headerCount
-                                         << " incomplete read diff=" << (msgSizeBytes - msgBufNext));
-                        }
-                    }
-                }
-            });
-            LOGS(_log, LOG_LVL_TRACE,
-                 context << " headerCount=" << headerCount << " msgSizeBytes=" << msgSizeBytes
-                         << " totalBytesRead=" << totalBytesRead);
-            if (msgSizeBufNext != 0) {
-                throw runtime_error("short read of the message header at offset " +
-                                    to_string(offset - msgSizeBytes) + ", file: " + httpUrl);
-            }
-            if (msgBufNext != 0) {
-                throw runtime_error("short read of the message body at offset " +
-                                    to_string(offset - msgSizeBytes) + ", file: " + httpUrl);
-            }
-        } catch (exception const& ex) {
-            LOGS(_log, LOG_LVL_ERROR, context + " " + ex.what());
-            success = false;
-    */
     }
 
     // Remove the file from the worker if it still exists. Report and ignore errors.
@@ -338,28 +236,20 @@ std::ostream& MergingHandler::print(std::ostream& os) const {
     return os << "MergingRequester(flushed=" << (_flushed ? "true)" : "false)");
 }
 
-bool queryIsNoLongerActive(qdisp::UberJob::Ptr const& uberJob) {  //&&&
-    // Do nothing if the query got cancelled for any reason.
-    if (uberJob->isQueryCancelled()) return true;
-
-    // Check for other indicators that the query may have cancelled or finished.
-    auto executive = uberJob->getExecutive();
-    if (executive == nullptr || executive->getCancelled() || executive->isRowLimitComplete()) {
-        return true;
-    }
-    return false;
-}
-
-bool MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uberJob, string const& fileUrl,
-                                uint64_t fileSize) {
+qdisp::MergeEndStatus MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uberJob, string const& fileUrl,
+                                                 uint64_t fileSize) {
     if (_flushed) {
         throw util::Bug(ERR_LOC, "already flushed");
     }
 
-    if (fileSize == 0) return true;
+    if (fileSize == 0) {
+        return qdisp::MergeEndStatus(true);
+    }
 
     // After this final test the job's result processing can't be interrupted.
-    if (uberJob->isQueryCancelled()) return true;
+    if (uberJob->isQueryCancelled()) {
+        return qdisp::MergeEndStatus(true);
+    }
 
     // Read from the http stream and push records into the CSV stream in a separate thread.
     // Note the fixed capacity of the stream which allows up to 2 records to be buffered
@@ -386,7 +276,7 @@ bool MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uberJob, string const
                 },
                 MergingHandler::_getHttpConnPool());
         // Push the stream terminator to indicate the end of the stream.
-        // It may be neeeded to unblock the table merger which may be still attempting to read
+        // It may be needed to unblock the table merger which may be still attempting to read
         // from the CSV stream.
         if (!fileReadErrorMsg.empty()) {
             csvStream->push(nullptr, 0);
@@ -407,61 +297,16 @@ bool MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uberJob, string const
         _setError(ccontrol::MSG_HTTP_RESULT, fileReadErrorMsg, util::ErrorCode::RESULT_IMPORT);
     }
     _flushed = true;
-    return fileMergeSuccess && fileReadErrorMsg.empty();
-}
-
-void MergingHandler::_setError(int code, std::string const& msg, int errorState) {
-    LOGS(_log, LOG_LVL_DEBUG, "_setError: code: " << code << ", message: " << msg);
-    auto exec = _executive.lock();
-    if (exec == nullptr) return;
-    exec->addMultiError(code, msg, errorState);
-}
-
-tuple<bool, bool> MergingHandler::flushHttp(string const& fileUrl, uint64_t fileSize, uint64_t expectedRows,
-                                            uint64_t& resultRows) {
-    bool success = false;
-    bool shouldCancel = false;
-
-    // This is needed to ensure the job query would be staying alive for the duration
-    // of the operation to prevent inconsistency within the application.
-    auto const uberJob = getUberJob().lock();
-    if (uberJob == nullptr) {
-        LOGS(_log, LOG_LVL_ERROR, __func__ << " failed, uberJob was NULL");
-        return {success, shouldCancel};  // both should still be false
-    }
 
-    LOGS(_log, LOG_LVL_TRACE,
-         "MergingHandler::" << __func__ << " uberJob=" << uberJob->getIdStr() << " fileUrl=" << fileUrl);
-
-    success = _mergeHttp(uberJob, fileUrl, fileSize);
-    // &&& FOULED_RESULTS   need to do something about shouldCancel.
-    // &&& until there is some way to know if csvStream has merged any bytes, just assume it has fouled the
-    // results.
-    if (!success) shouldCancel = true;
-
-    if (!success || shouldCancel) {
-        LOGS(_log, LOG_LVL_WARN, __func__ << " success=" << success << " shouldCancel=" << shouldCancel);
-    }
-
-    if (success) {
-        _infileMerger->mergeCompleteFor(uberJob->getUjId());
-    }
-    return {success, shouldCancel};
-}
-
-/* &&&
-bool MergingHandler::_mergeHttp(shared_ptr<qdisp::UberJob> const& uberJob,
-                                proto::ResponseData const& responseData) {
-    if (_flushed) {
-        throw util::Bug(ERR_LOC, "already flushed");
-    }
-    bool const success = _infileMerger->mergeHttp(uberJob, responseData);
-    if (!success) {
-        LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
-        util::Error const& err = _infileMerger->getError();
-        _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg(), util::ErrorCode::RESULT_IMPORT);
+    qdisp::MergeEndStatus mergeEStatus(fileMergeSuccess && fileReadErrorMsg.empty());
+    if (!mergeEStatus.success) {
+        // This error check needs to come after the csvThread.join() to avoid race conditions.
+        if (csvStream->getBytesWritten() > 0) {
+            // There was a failure and bytes were written, result table is ruined.
+            mergeEStatus.contaminated = true;
+        }
     }
-    return success;
+    return mergeEStatus;
 }
 
 void MergingHandler::_setError(int code, std::string const& msg, int errorState) {
@@ -471,54 +316,25 @@ void MergingHandler::_setError(int code, std::string const& msg, int errorState)
     exec->addMultiError(code, msg, errorState);
 }
 
-tuple<bool, bool> MergingHandler::flushHttp(string const& fileUrl, uint64_t expectedRows,
-                                            uint64_t& resultRows) {
-    bool success = false;
-    bool shouldCancel = false;
-
+qdisp::MergeEndStatus MergingHandler::flushHttp(string const& fileUrl, uint64_t fileSize) {
     // This is needed to ensure the job query would be staying alive for the duration
     // of the operation to prevent inconsistency within the application.
     auto const uberJob = getUberJob().lock();
     if (uberJob == nullptr) {
         LOGS(_log, LOG_LVL_ERROR, __func__ << " failed, uberJob was NULL");
-        return {success, shouldCancel};  // both should still be false
+        return qdisp::MergeEndStatus(false);
     }
 
     LOGS(_log, LOG_LVL_TRACE,
          "MergingHandler::" << __func__ << " uberJob=" << uberJob->getIdStr() << " fileUrl=" << fileUrl);
 
-    // Dispatch result processing to the corresponidng method which depends on
-    // the result delivery protocol configured at the worker.
-    // Notify the file reader when all rows have been read by setting 'last = true'.
-    auto const dataMergerHttp = [&](char const* buf, uint32_t bufSize, bool& last) {
-        LOGS(_log, LOG_LVL_TRACE, "dataMergerHttp");
-        last = true;
-        proto::ResponseData responseData;
-        if (responseData.ParseFromArray(buf, bufSize) && responseData.IsInitialized()) {
-            bool const mergeSuccess = _mergeHttp(uberJob, responseData);
-            if (mergeSuccess) {
-                resultRows += responseData.row_size();
-                last = resultRows >= expectedRows;
-            }
-            return mergeSuccess;
-        }
-        throw runtime_error("MergingHandler::flush ** message deserialization failed **");
-    };
-
-    tie(success, shouldCancel) =
-            ::readHttpFileAndMergeHttp(uberJob, fileUrl, dataMergerHttp, MergingHandler::_getHttpConnPool());
-
-    if (!success || shouldCancel) {
-        LOGS(_log, LOG_LVL_WARN, __func__ << " success=" << success << " shouldCancel=" << shouldCancel);
-    }
-
-    if (success) {
+    qdisp::MergeEndStatus mergeStatus = _mergeHttp(uberJob, fileUrl, fileSize);
+    if (mergeStatus.success) {
         _infileMerger->mergeCompleteFor(uberJob->getUjId());
     }
-    return {success, shouldCancel};
+    return mergeStatus;
 }
 
->>>>>>> a27525c04017db9a30061fa0bb4b5228c0c5d1b2   */
 void MergingHandler::flushHttpError(int errorCode, std::string const& errorMsg, int errState) {
     if (!_errorSet.exchange(true)) {
         _setError(errorCode, errorMsg, errState);

src/ccontrol/MergingHandler.h

@@ -32,15 +32,14 @@
 #include "qdisp/ResponseHandler.h"
 
 // Forward declarations
-
-namespace lsst::qserv::proto {
-class ResponseData;
-class ResponseSummary;
-}  // namespace lsst::qserv::proto
+namespace lsst::qserv::http {
+class ClientConnPool;
+}  // namespace lsst::qserv::http
 
 namespace lsst::qserv::qdisp {
 class Executive;
 class JobQuery;
+class MergeEndStatus;
 class UberJob;
 }  // namespace lsst::qserv::qdisp
 
@@ -70,12 +69,8 @@ class MergingHandler : public qdisp::ResponseHandler {
 
     /// @see ResponseHandler::flushHttp
     /// @see MerginHandler::_mergeHttp
-    std::tuple<bool, bool> flushHttp(
-            std::string const& fileUrl, uint64_t fileSize, uint64_t expectedRows,
-            /* &&&
-                std::tuple<bool, bool> flushHttp(std::string const& fileUrl, uint64_t expectedRows,
-            >>>>>>> a27525c04017db9a30061fa0bb4b5228c0c5d1b2 */
-            uint64_t& resultRows) override;
+    /// @see qdisp::MergeEndStatus
+    qdisp::MergeEndStatus flushHttp(std::string const& fileUrl, uint64_t fileSize) override;
 
     /// @see ResponseHandler::flushHttpError
     void flushHttpError(int errorCode, std::string const& errorMsg, int status) override;
@@ -88,9 +83,8 @@ class MergingHandler : public qdisp::ResponseHandler {
 
 private:
     /// Call InfileMerger to do the work of merging this data to the result.
-
-    bool _mergeHttp(std::shared_ptr<qdisp::UberJob> const& uberJob, std::string const& fileUrl,
-                    uint64_t fileSize);
+    qdisp::MergeEndStatus _mergeHttp(std::shared_ptr<qdisp::UberJob> const& uberJob,
+                                     std::string const& fileUrl, uint64_t fileSize);
 
     /// Set error code and string.
     void _setError(int code, std::string const& msg, int errorState);

src/ccontrol/UserQuerySelect.cc

@@ -84,7 +84,6 @@
 #include "czar/CzarRegistry.h"
 #include "global/constants.h"
 #include "global/LogContext.h"
-#include "proto/worker.pb.h"
 #include "qdisp/Executive.h"
 #include "qdisp/JobQuery.h"
 #include "qmeta/MessageStore.h"

src/czar/CMakeLists.txt

@@ -1,5 +1,4 @@
 add_library(czar SHARED)
-add_dependencies(czar proto)
 
 target_sources(czar PRIVATE
     ActiveWorker.cc