text:
// Copyright 2016 The Streamer Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "gige_file_writer.h"

#include <boost/filesystem.hpp>

GigeFileWriter::GigeFileWriter(const std::string& directory,
                               size_t frames_per_file)
    : Processor(PROCESSOR_TYPE_CUSTOM, {"input"}, {}),
      directory_(directory),
      frames_written_(0),
      frames_per_file_(frames_per_file) {}

bool GigeFileWriter::Init() {
  // Make a directory for this run
  if (!boost::filesystem::exists(directory_)) {
    boost::filesystem::create_directory(directory_);
  } else {
    LOG(WARNING) << "Directory: " << directory_
                 << " already exists, may re-write existed files";
  }

  frames_written_ = 0;
  current_filename_ = "";

  return true;
}

bool GigeFileWriter::OnStop() {
  if (current_file_.is_open()) {
    current_file_.close();
  }
  return true;
}

void GigeFileWriter::Process() {
  // Create file
  if (frames_written_ % frames_per_file_ == 0) {
    std::ostringstream ss;
    ss << frames_written_ / frames_per_file_ << ".dat";

    std::string filename = directory_ + "/" + ss.str();

    if (current_file_.is_open()) current_file_.close();

    current_file_.open(filename, std::ios::binary | std::ios::out);

    if (!current_file_.is_open()) {
      LOG(FATAL) << "Can't open file: " << filename << " for write";
    }

    current_filename_ = filename;
  }

  auto frame = GetFrame("input");

  std::vector<char> raw_pixels =
      frame->GetValue<std::vector<char>>("original_bytes");
  current_file_.write((char*)raw_pixels.data(), raw_pixels.size());

  frames_written_ += 1;
}

max_stars_repo_name: GTkernel/streamer

max_issues_repo_name: GTkernel/streamer

max_forks_repo_name: GTkernel/streamer


text:
/**
 * Copyright Soramitsu Co., Ltd. All Rights Reserved.
 * SPDX-License-Identifier: Apache-2.0
 */

#include "application/impl/local_key_storage.hpp"

#include <boost/property_tree/json_parser.hpp>
#include "application/impl/config_reader/error.hpp"
#include "application/impl/config_reader/pt_util.hpp"
#include "common/hexutil.hpp"

namespace kagome::application {

  outcome::result<std::shared_ptr<LocalKeyStorage>> LocalKeyStorage::create(
      const std::string &keystore_path) {
    auto storage = LocalKeyStorage();
    OUTCOME_TRY(storage.loadFromJson(keystore_path));
    return std::make_shared<LocalKeyStorage>(std::move(storage));
  }

  kagome::crypto::SR25519Keypair LocalKeyStorage::getLocalSr25519Keypair()
      const {
    return sr_25519_keypair_;
  }

  kagome::crypto::ED25519Keypair LocalKeyStorage::getLocalEd25519Keypair()
      const {
    return ed_25519_keypair_;
  }

  libp2p::crypto::KeyPair LocalKeyStorage::getP2PKeypair() const {
    return p2p_keypair_;
  }

  outcome::result<void> LocalKeyStorage::loadFromJson(const std::string &file) {
    namespace pt = boost::property_tree;
    pt::ptree tree;
    try {
      pt::read_json(file, tree);
    } catch (pt::json_parser_error &e) {
      return ConfigReaderError::PARSER_ERROR;
    }
    OUTCOME_TRY(loadSR25519Keys(tree));
    OUTCOME_TRY(loadED25519Keys(tree));
    OUTCOME_TRY(loadP2PKeys(tree));

    return outcome::success();
  }

  outcome::result<void> LocalKeyStorage::loadSR25519Keys(
      const boost::property_tree::ptree &tree) {
    OUTCOME_TRY(sr_tree, ensure(tree.get_child_optional("sr25519keypair")));

    OUTCOME_TRY(sr_pubkey_str,
                ensure(sr_tree.get_optional<std::string>("public")));
    OUTCOME_TRY(sr_privkey_str,
                ensure(sr_tree.get_optional<std::string>("private")));

    // get rid of 0x from beginning
    OUTCOME_TRY(sr_pubkey_buffer, common::unhexWith0x(sr_pubkey_str));
    OUTCOME_TRY(sr_privkey_buffer, common::unhexWith0x(sr_privkey_str));

    OUTCOME_TRY(sr_pubkey,
                crypto::SR25519PublicKey::fromSpan(sr_pubkey_buffer));
    OUTCOME_TRY(sr_privkey,
                crypto::SR25519SecretKey::fromSpan(sr_privkey_buffer));

    sr_25519_keypair_.public_key = sr_pubkey;
    sr_25519_keypair_.secret_key = sr_privkey;
    return outcome::success();
  }

  outcome::result<void> LocalKeyStorage::loadED25519Keys(
      const boost::property_tree::ptree &tree) {
    auto ed_tree_opt = tree.get_child_optional("ed25519keypair");
    if (not ed_tree_opt) return ConfigReaderError::MISSING_ENTRY;
    const auto &ed_tree = ed_tree_opt.value();

    OUTCOME_TRY(ed_pubkey_str,
                ensure(ed_tree.get_optional<std::string>("public")));
    OUTCOME_TRY(ed_privkey_str,
                ensure(ed_tree.get_optional<std::string>("private")));

    // get rid of 0x from beginning
    OUTCOME_TRY(ed_pubkey_buffer, common::unhexWith0x(ed_pubkey_str));
    OUTCOME_TRY(ed_privkey_buffer, common::unhexWith0x(ed_privkey_str));

    OUTCOME_TRY(ed_pubkey,
                crypto::ED25519PublicKey::fromSpan(ed_pubkey_buffer));
    OUTCOME_TRY(ed_privkey,
                crypto::ED25519PrivateKey::fromSpan(ed_privkey_buffer));

    ed_25519_keypair_.public_key = ed_pubkey;
    ed_25519_keypair_.private_key = ed_privkey;
    return outcome::success();
  }

  outcome::result<void> LocalKeyStorage::loadP2PKeys(
      const boost::property_tree::ptree &tree) {
    auto p2p_tree_opt = tree.get_child_optional("p2p_keypair");
    if (not p2p_tree_opt) return ConfigReaderError::MISSING_ENTRY;
    const auto &p2p_tree = p2p_tree_opt.value();

    OUTCOME_TRY(p2p_type,
                ensure(p2p_tree.get_optional<std::string>("p2p_type")));

    using KeyType = libp2p::crypto::Key::Type;
    if (p2p_type == "ed25519") {
      p2p_keypair_.publicKey.type = KeyType::Ed25519;
      p2p_keypair_.privateKey.type = KeyType::Ed25519;
    } else if (p2p_type == "rsa") {
      p2p_keypair_.publicKey.type = KeyType::RSA;
      p2p_keypair_.privateKey.type = KeyType::RSA;
    } else if (p2p_type == "secp256k1") {
      p2p_keypair_.publicKey.type = KeyType::Secp256k1;
      p2p_keypair_.privateKey.type = KeyType::Secp256k1;
    } else if (p2p_type == "ecdsa") {
      p2p_keypair_.publicKey.type = KeyType::ECDSA;
      p2p_keypair_.privateKey.type = KeyType::ECDSA;
    } else {
      p2p_keypair_.publicKey.type = KeyType::UNSPECIFIED;
      p2p_keypair_.privateKey.type = KeyType::UNSPECIFIED;
    }

    auto p2p_keypair_tree_opt = p2p_tree.get_child_optional("keypair");
    if (not p2p_keypair_tree_opt) return ConfigReaderError::MISSING_ENTRY;
    const auto &p2p_keypair_tree = p2p_keypair_tree_opt.value();

    OUTCOME_TRY(p2p_public_key_str,
                ensure(p2p_keypair_tree.get_optional<std::string>("public")));
    OUTCOME_TRY(p2p_private_key_str,
                ensure(p2p_keypair_tree.get_optional<std::string>("private")));

    // get rid of 0x from beginning
    OUTCOME_TRY(p2p_public_key, common::unhexWith0x(p2p_public_key_str));
    OUTCOME_TRY(p2p_private_key, common::unhexWith0x(p2p_private_key_str));

    p2p_keypair_.publicKey.data = p2p_public_key;
    p2p_keypair_.privateKey.data = p2p_private_key;

    return outcome::success();
  }
}  // namespace kagome::application

ext: cpp

max_stars_repo_name: FlorianFranzen/kagome

max_issues_repo_name: FlorianFranzen/kagome

max_forks_repo_name: FlorianFranzen/kagome


text:
/**
 *    Copyright (C) 2021-present MongoDB, Inc.
 *
 *    This program is free software: you can redistribute it and/or modify
 *    it under the terms of the Server Side Public License, version 1,
 *    as published by MongoDB, Inc.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    Server Side Public License for more details.
 *
 *    You should have received a copy of the Server Side Public License
 *    along with this program. If not, see
 *    <http://www.mongodb.com/licensing/server-side-public-license>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the Server Side Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#include "mongo/platform/basic.h"

#include <boost/optional/optional_io.hpp>
#include <vector>

#include "mongo/db/catalog_raii.h"
#include "mongo/db/exec/document_value/document.h"
#include "mongo/db/logical_session_id.h"
#include "mongo/db/logical_session_id_helpers.h"
#include "mongo/db/persistent_task_store.h"
#include "mongo/db/repl/replication_coordinator_mock.h"
#include "mongo/db/repl/session_update_tracker.h"
#include "mongo/db/repl/storage_interface_impl.h"
#include "mongo/db/s/resharding/donor_oplog_id_gen.h"
#include "mongo/db/s/resharding/resharding_oplog_session_application.h"
#include "mongo/db/service_context_d_test_fixture.h"
#include "mongo/db/session_catalog_mongod.h"
#include "mongo/db/transaction_participant.h"
#include "mongo/unittest/death_test.h"
#include "mongo/unittest/unittest.h"

#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest

namespace mongo {
namespace {

class ReshardingOplogSessionApplicationTest : public ServiceContextMongoDTest {
public:
    void setUp() override {
        ServiceContextMongoDTest::setUp();

        auto serviceContext = getServiceContext();
        {
            auto opCtx = makeOperationContext();
            auto replCoord = std::make_unique<repl::ReplicationCoordinatorMock>(serviceContext);
            ASSERT_OK(replCoord->setFollowerMode(repl::MemberState::RS_PRIMARY));
            repl::ReplicationCoordinator::set(serviceContext, std::move(replCoord));

            repl::createOplog(opCtx.get());

            auto storageImpl = std::make_unique<repl::StorageInterfaceImpl>();
            repl::StorageInterface::set(serviceContext, std::move(storageImpl));

            MongoDSessionCatalog::onStepUp(opCtx.get());
        }

        serverGlobalParams.clusterRole = ClusterRole::ShardServer;
    }

    repl::OpTime insertSessionRecord(OperationContext* opCtx,
                                     LogicalSessionId lsid,
                                     TxnNumber txnNumber,
                                     std::vector<StmtId> stmtIds) {
        opCtx->setLogicalSessionId(std::move(lsid));
        opCtx->setTxnNumber(txnNumber);

        MongoDOperationContextSession ocs(opCtx);
        auto txnParticipant = TransactionParticipant::get(opCtx);
        txnParticipant.beginOrContinue(
            opCtx, {txnNumber}, boost::none /* autocommit */, boost::none /* startTransaction */);

        auto opTime = [opCtx] {
            WriteUnitOfWork wuow(opCtx);
            ScopeGuard guard{[&wuow] { wuow.commit(); }};
            return repl::getNextOpTime(opCtx);
        }();
        WriteUnitOfWork wuow(opCtx);
        SessionTxnRecord sessionTxnRecord(*opCtx->getLogicalSessionId(),
                                          *opCtx->getTxnNumber(),
                                          opTime,
                                          opCtx->getServiceContext()->getFastClockSource()->now());
        txnParticipant.onWriteOpCompletedOnPrimary(opCtx, std::move(stmtIds), sessionTxnRecord);
        wuow.commit();

        return opTime;
    }

    void insertOp(OperationContext* opCtx, const BSONObj& oplogBson) {
        DBDirectClient client(opCtx);
        client.insert(_oplogBufferNss.toString(), oplogBson);
    }

    void makeInProgressTxn(OperationContext* opCtx, LogicalSessionId lsid, TxnNumber txnNumber) {
        opCtx->setInMultiDocumentTransaction();
        opCtx->setLogicalSessionId(std::move(lsid));
        opCtx->setTxnNumber(txnNumber);

        MongoDOperationContextSession ocs(opCtx);
        auto txnParticipant = TransactionParticipant::get(opCtx);
        txnParticipant.beginOrContinue(
            opCtx, {txnNumber}, false /* autocommit */, true /* startTransaction */);

        txnParticipant.unstashTransactionResources(opCtx, "insert");
        txnParticipant.stashTransactionResources(opCtx);
    }

    repl::OpTime makePreparedTxn(OperationContext* opCtx,
                                 LogicalSessionId lsid,
                                 TxnNumber txnNumber) {
        opCtx->setInMultiDocumentTransaction();
        opCtx->setLogicalSessionId(std::move(lsid));
        opCtx->setTxnNumber(txnNumber);

        MongoDOperationContextSession ocs(opCtx);
        auto txnParticipant = TransactionParticipant::get(opCtx);
        txnParticipant.beginOrContinue(
            opCtx, {txnNumber}, false /* autocommit */, true /* startTransaction */);

        txnParticipant.unstashTransactionResources(opCtx, "prepareTransaction");

        // The transaction machinery cannot store an empty locker.
        { Lock::GlobalLock globalLock(opCtx, MODE_IX); }
        auto opTime = [opCtx] {
            TransactionParticipant::SideTransactionBlock sideTxn{opCtx};

            WriteUnitOfWork wuow{opCtx};
            auto opTime = repl::getNextOpTime(opCtx);
            wuow.release();

            opCtx->recoveryUnit()->abortUnitOfWork();
            opCtx->lockState()->endWriteUnitOfWork();

            return opTime;
        }();
        txnParticipant.prepareTransaction(opCtx, opTime);
        txnParticipant.stashTransactionResources(opCtx);

        return opTime;
    }

    void abortTxn(OperationContext* opCtx, LogicalSessionId lsid, TxnNumber txnNumber) {
        opCtx->setInMultiDocumentTransaction();
        opCtx->setLogicalSessionId(std::move(lsid));
        opCtx->setTxnNumber(txnNumber);

        MongoDOperationContextSession ocs(opCtx);
        auto txnParticipant = TransactionParticipant::get(opCtx);
        txnParticipant.beginOrContinue(
            opCtx, {txnNumber}, false /* autocommit */, boost::none /* startTransaction */);

        txnParticipant.unstashTransactionResources(opCtx, "abortTransaction");
        txnParticipant.abortTransaction(opCtx);
        txnParticipant.stashTransactionResources(opCtx);
    }

    repl::OplogEntry makeNoopOp(BSONObj object,
                                LogicalSessionId lsid,
                                TxnNumber txnNumber,
                                repl::OpTime opTime) {
        repl::MutableOplogEntry op;
        op.setOpType(repl::OpTypeEnum::kNoop);
        op.setObject(object);
        op.setSessionId(std::move(lsid));
        op.setTxnNumber(std::move(txnNumber));
        op.setOpTime(opTime);
        op.set_id(
            Value{Document{{ReshardingDonorOplogId::kClusterTimeFieldName, opTime.getTimestamp()},
                           {ReshardingDonorOplogId::kTsFieldName, opTime.getTimestamp()}}});

        // These are unused by ReshardingOplogSessionApplication but required by IDL parsing.
        op.setNss({});
        op.setWallClockTime({});

        return {op.toBSON()};
    }

    repl::OplogEntry makeUpdateOp(
        BSONObj document,
        LogicalSessionId lsid,
        TxnNumber txnNumber,
        const std::vector<StmtId>& stmtIds,
        boost::optional<repl::RetryImageEnum> needsRetryImage = boost::none,
        boost::optional<repl::OpTime> preImageOpTime = boost::none,
        boost::optional<repl::OpTime> postImageOpTime = boost::none) {
        repl::MutableOplogEntry op;
        op.setOpType(repl::OpTypeEnum::kUpdate);
        op.setObject2(document["_id"].wrap().getOwned());
        op.setObject(std::move(document));
        op.setSessionId(std::move(lsid));
        op.setTxnNumber(std::move(txnNumber));
        op.setStatementIds(stmtIds);
        op.setNeedsRetryImage(needsRetryImage);
        op.setPreImageOpTime(preImageOpTime);
        op.setPostImageOpTime(postImageOpTime);
        op.setOpTime({{}, {}});
        op.set_id(Value{
            Document{{ReshardingDonorOplogId::kClusterTimeFieldName, op.getOpTime().getTimestamp()},
                     {ReshardingDonorOplogId::kTsFieldName, op.getOpTime().getTimestamp()}}});

        // These are unused by ReshardingOplogSessionApplication but required by IDL parsing.
        op.setNss({});
        op.setWallClockTime({});

        return {op.toBSON()};
    }

    repl::OplogEntry makeFinishTxnOp(LogicalSessionId lsid, TxnNumber txnNumber) {
        repl::MutableOplogEntry op;
        op.setOpType(repl::OpTypeEnum::kCommand);
        // Use AbortTransactionOplogObject rather than CommitTransactionOplogObject to avoid needing
        // to deal with the commitTimestamp value. Both are treated the same way by
        // ReshardingOplogSessionApplication anyway.
        op.setObject(AbortTransactionOplogObject{}.toBSON());
        op.setSessionId(std::move(lsid));
        op.setTxnNumber(std::move(txnNumber));
        op.setOpTime({{}, {}});
        op.set_id(Value{
            Document{{ReshardingDonorOplogId::kClusterTimeFieldName, op.getOpTime().getTimestamp()},
                     {ReshardingDonorOplogId::kTsFieldName, op.getOpTime().getTimestamp()}}});

        // These are unused by ReshardingOplogSessionApplication but required by IDL parsing.
        op.setNss({});
        op.setWallClockTime({});

        return {op.toBSON()};
    }

    size_t getNumOplogEntries(OperationContext* opCtx) {
        PersistentTaskStore<repl::OplogEntryBase> store(NamespaceString::kRsOplogNamespace);
        return store.count(opCtx);
    }

    std::vector<repl::DurableOplogEntry> findOplogEntriesNewerThan(OperationContext* opCtx,
                                                                   Timestamp ts) {
        std::vector<repl::DurableOplogEntry> result;

        PersistentTaskStore<repl::OplogEntryBase> store(NamespaceString::kRsOplogNamespace);
        store.forEach(opCtx, BSON("ts" << BSON("$gt" << ts)), [&](const auto& oplogEntry) {
            result.emplace_back(
                unittest::assertGet(repl::DurableOplogEntry::parse(oplogEntry.toBSON())));
            return true;
        });

        return result;
    }

    boost::optional<SessionTxnRecord> findSessionRecord(OperationContext* opCtx,
                                                        const LogicalSessionId& lsid) {
        boost::optional<SessionTxnRecord> result;

        PersistentTaskStore<SessionTxnRecord> store(
            NamespaceString::kSessionTransactionsTableNamespace);
        store.forEach(opCtx,
                      BSON(SessionTxnRecord::kSessionIdFieldName << lsid.toBSON()),
                      [&](const auto& sessionTxnRecord) {
                          result.emplace(sessionTxnRecord);
                          return false;
                      });

        return result;
    }

    void checkGeneratedNoop(const repl::DurableOplogEntry& foundOp,
                            const LogicalSessionId& lsid,
                            TxnNumber txnNumber,
                            const std::vector<StmtId>& stmtIds) {
        ASSERT_EQ(OpType_serializer(foundOp.getOpType()),
                  OpType_serializer(repl::OpTypeEnum::kNoop))
            << foundOp;

        ASSERT_EQ(foundOp.getSessionId(), lsid) << foundOp;
        ASSERT_EQ(foundOp.getTxnNumber(), txnNumber) << foundOp;
        ASSERT(foundOp.getStatementIds() == stmtIds) << foundOp;

        // The oplog entry must have o2 and fromMigrate set or SessionUpdateTracker will ignore it.
        ASSERT_TRUE(foundOp.getObject2());
        ASSERT_TRUE(foundOp.getFromMigrate());
    }

    void checkSessionTxnRecord(const SessionTxnRecord& sessionTxnRecord,
                               const repl::DurableOplogEntry& foundOp) {
        ASSERT_EQ(sessionTxnRecord.getSessionId(), foundOp.getSessionId())
            << sessionTxnRecord.toBSON() << ", " << foundOp;
        ASSERT_EQ(sessionTxnRecord.getTxnNum(), foundOp.getTxnNumber())
            << sessionTxnRecord.toBSON() << ", " << foundOp;
        ASSERT_EQ(sessionTxnRecord.getLastWriteOpTime(), foundOp.getOpTime())
            << sessionTxnRecord.toBSON() << ", " << foundOp;
        ASSERT_EQ(sessionTxnRecord.getLastWriteDate(), foundOp.getWallClockTime())
            << sessionTxnRecord.toBSON() << ", " << foundOp;

        // Verify secondaries during replication's oplog application would generate an identical
        // SessionTxnRecord update from the oplog entry.
        repl::SessionUpdateTracker sessionUpdateTracker;
        auto flushImmediate = sessionUpdateTracker.updateSession({foundOp.toBSON()});
        auto flushLater = sessionUpdateTracker.flushAll();

        // The oplog entries generated by ReshardingTxnCloner are always op='n' and are therefore
        // buffered by SessionUpdateTracker rather than flushed immediately.
        ASSERT_FALSE(bool(flushImmediate));
        ASSERT_EQ(flushLater.size(), 1U);

        ASSERT_EQ(OpType_serializer(flushLater[0].getOpType()),
                  OpType_serializer(repl::OpTypeEnum::kUpdate))
            << flushLater[0].getEntry();
        ASSERT_BSONOBJ_BINARY_EQ(*flushLater[0].getObject2(),
                                 BSON("_id" << sessionTxnRecord.getSessionId().toBSON()));
        ASSERT_BSONOBJ_BINARY_EQ(flushLater[0].getObject(), sessionTxnRecord.toBSON());
    }

    void checkStatementExecuted(OperationContext* opCtx,
                                LogicalSessionId lsid,
                                TxnNumber txnNumber,
                                StmtId stmtId) {
        opCtx->setLogicalSessionId(std::move(lsid));
        opCtx->setTxnNumber(txnNumber);

        MongoDOperationContextSession ocs(opCtx);
        auto txnParticipant = TransactionParticipant::get(opCtx);
        txnParticipant.beginOrContinue(
            opCtx, {txnNumber}, boost::none /* autocommit */, boost::none /* startTransaction */);
        ASSERT_TRUE(bool(txnParticipant.checkStatementExecuted(opCtx, stmtId)));
    }

    const NamespaceString& oplogBufferNss() {
        return _oplogBufferNss;
    }

private:
    // Used for pre/post image oplog entry lookup.
    const ShardId _donorShardId{"donor-0"};
    const NamespaceString _oplogBufferNss{NamespaceString::kReshardingLocalOplogBufferPrefix +
                                          _donorShardId.toString()};
};

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteForNewSession) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, incomingTxnNumber, {incomingStmtId});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {incomingStmtId});

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteForNewSessionMultiStmtIds) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 1;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3, 4});
    }();

    auto oplogEntry = makeUpdateOp(
        BSON("_id" << 1), lsid, incomingTxnNumber, {incomingStmtId, incomingStmtId + 1});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(
            foundOps[0], lsid, incomingTxnNumber, {incomingStmtId, incomingStmtId + 1});

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId + 1);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteHasHigherTxnNumber) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber existingTxnNumber = 100;
    StmtId existingStmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, existingTxnNumber, {existingStmtId});
    }();

    TxnNumber incomingTxnNumber = existingTxnNumber + 1;
    StmtId incomingStmtId = 2;
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, incomingTxnNumber, {incomingStmtId});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {incomingStmtId});

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
        ASSERT_THROWS_CODE(
            checkStatementExecuted(opCtx.get(), lsid, existingTxnNumber, existingStmtId),
            DBException,
            ErrorCodes::TransactionTooOld);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteHasLowerTxnNumber) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber existingTxnNumber = 100;
    StmtId existingStmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, existingTxnNumber, {existingStmtId});
    }();

    TxnNumber incomingTxnNumber = existingTxnNumber - 1;
    StmtId incomingStmtId = 2;
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, incomingTxnNumber, {incomingStmtId});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        ASSERT_EQ(sessionTxnRecord->getSessionId(), lsid) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getTxnNum(), existingTxnNumber) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getLastWriteOpTime(), opTime) << sessionTxnRecord->toBSON();
    }

    {
        auto opCtx = makeOperationContext();
        ASSERT_THROWS_CODE(
            checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId),
            DBException,
            ErrorCodes::TransactionTooOld);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest,
       IncomingRetryableWriteHasEqualTxnNumberAsRetryableWrite) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber txnNumber = 100;
    StmtId existingStmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, txnNumber, {existingStmtId});
    }();

    StmtId incomingStmtId = 2;
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, txnNumber, {incomingStmtId});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, txnNumber, {incomingStmtId});
        ASSERT_EQ(foundOps[0].getPrevWriteOpTimeInTransaction(), opTime) << foundOps[0];

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, txnNumber, incomingStmtId);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteHasEqualTxnNumberAsTxn) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber txnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        auto opTime = makePreparedTxn(opCtx.get(), lsid, txnNumber);
        abortTxn(opCtx.get(), lsid, txnNumber);
        return opTime;
    }();

    StmtId incomingStmtId = 2;
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, txnNumber, {incomingStmtId});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteStmtAlreadyExecuted) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber txnNumber = 100;
    StmtId stmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, txnNumber, {stmtId});
    }();

    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, txnNumber, {stmtId});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        ASSERT_EQ(sessionTxnRecord->getSessionId(), lsid) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getTxnNum(), txnNumber) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getLastWriteOpTime(), opTime) << sessionTxnRecord->toBSON();
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteStmtsAlreadyExecuted) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber txnNumber = 100;
    StmtId stmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, txnNumber, {stmtId, stmtId + 1});
    }();

    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, txnNumber, {stmtId, stmtId + 1});

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        ASSERT_EQ(sessionTxnRecord->getSessionId(), lsid) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getTxnNum(), txnNumber) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getLastWriteOpTime(), opTime) << sessionTxnRecord->toBSON();
    }
}

TEST_F(ReshardingOplogSessionApplicationTest,
       IncomingRetryableWriteHasHigherTxnNumberThanPreparedTxn) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber existingTxnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return makePreparedTxn(opCtx.get(), lsid, existingTxnNumber);
    }();

    TxnNumber incomingTxnNumber = existingTxnNumber + 1;
    StmtId incomingStmtId = 2;
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1), lsid, incomingTxnNumber, {incomingStmtId});

    auto conflictingTxnCompletionFuture = [&] {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        return applier.tryApplyOperation(opCtx.get(), oplogEntry);
    }();

    ASSERT_TRUE(bool(conflictingTxnCompletionFuture));
    ASSERT_FALSE(conflictingTxnCompletionFuture->isReady());

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }

    {
        auto opCtx = makeOperationContext();
        abortTxn(opCtx.get(), lsid, existingTxnNumber);
    }

    ASSERT_TRUE(conflictingTxnCompletionFuture->isReady());
    ASSERT_OK(conflictingTxnCompletionFuture->getNoThrow());
}

TEST_F(ReshardingOplogSessionApplicationTest,
       IncomingRetryableWriteHasInProgressRetryableInternalTransaction) {
    auto retryableWriteLsid = makeLogicalSessionIdForTest();
    TxnNumber retryableWriteTxnNumber = 100;
    auto internalTxnLsid = makeLogicalSessionIdWithTxnNumberAndUUIDForTest(retryableWriteLsid,
                                                                           retryableWriteTxnNumber);
    TxnNumber internalTxnTxnNumber = 1;
    StmtId stmtId = 2;

    {
        auto opCtx = makeOperationContext();
        makeInProgressTxn(opCtx.get(), internalTxnLsid, internalTxnTxnNumber);
    }

    auto oplogEntry =
        makeUpdateOp(BSON("_id" << 1), retryableWriteLsid, retryableWriteTxnNumber, {stmtId});

    auto conflictingTxnCompletionFuture = [&] {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        return applier.tryApplyOperation(opCtx.get(), oplogEntry);
    }();

    ASSERT_TRUE(bool(conflictingTxnCompletionFuture));
    ASSERT_FALSE(conflictingTxnCompletionFuture->isReady());

    {
        auto opCtx = makeOperationContext();
        ASSERT_EQ(getNumOplogEntries(opCtx.get()), 0U);
        auto sessionTxnRecord = findSessionRecord(opCtx.get(), retryableWriteLsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }

    {
        auto opCtx = makeOperationContext();
        abortTxn(opCtx.get(), internalTxnLsid, internalTxnTxnNumber);
    }

    ASSERT_TRUE(conflictingTxnCompletionFuture->isReady());
    ASSERT_OK(conflictingTxnCompletionFuture->getNoThrow());
}

TEST_F(ReshardingOplogSessionApplicationTest,
       IncomingRetryableWriteHasPreparedRetryableInternalTransaction) {
    auto retryableWriteLsid = makeLogicalSessionIdForTest();
    TxnNumber retryableWriteTxnNumber = 100;
    auto internalTxnLsid = makeLogicalSessionIdWithTxnNumberAndUUIDForTest(retryableWriteLsid,
                                                                           retryableWriteTxnNumber);
    TxnNumber internalTxnTxnNumber = 1;
    StmtId stmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return makePreparedTxn(opCtx.get(), internalTxnLsid, internalTxnTxnNumber);
    }();

    auto oplogEntry =
        makeUpdateOp(BSON("_id" << 1), retryableWriteLsid, retryableWriteTxnNumber, {stmtId});

    auto conflictingTxnCompletionFuture = [&] {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        return applier.tryApplyOperation(opCtx.get(), oplogEntry);
    }();

    ASSERT_TRUE(bool(conflictingTxnCompletionFuture));
    ASSERT_FALSE(conflictingTxnCompletionFuture->isReady());

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), retryableWriteLsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }

    {
        auto opCtx = makeOperationContext();
        abortTxn(opCtx.get(), internalTxnLsid, internalTxnTxnNumber);
    }

    ASSERT_TRUE(conflictingTxnCompletionFuture->isReady());
    ASSERT_OK(conflictingTxnCompletionFuture->getNoThrow());
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteHasPreImage) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    auto preImageOplogEntry =
        makeNoopOp(BSON("_id" << 1 << "preImage" << true), lsid, incomingTxnNumber, {{1, 0}, 1});
    {
        auto opCtx = makeOperationContext();
        insertOp(opCtx.get(), preImageOplogEntry.getEntry().toBSON());
    }
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1),
                                   lsid,
                                   incomingTxnNumber,
                                   {incomingStmtId},
                                   boost::none /* needsRetryImage */,
                                   preImageOplogEntry.getOpTime() /* preImageOpTime */);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 2U);
        ASSERT_BSONOBJ_BINARY_EQ(foundOps[0].getObject(), BSON("_id" << 1 << "preImage" << true));
        checkGeneratedNoop(foundOps[1], lsid, incomingTxnNumber, {incomingStmtId});
        ASSERT_EQ(foundOps[1].getPreImageOpTime(), foundOps[0].getOpTime()) << foundOps[1];

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[1]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
    }
}

DEATH_TEST_REGEX_F(ReshardingOplogSessionApplicationTest,
                   IncomingRetryableWriteHasPreImageThatCannotBeFound,
                   "Tripwire assertion.*6344401") {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    repl::OpTime preImageOpTime{{1, 0}, 1};
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1),
                                   lsid,
                                   incomingTxnNumber,
                                   {incomingStmtId},
                                   boost::none /* needsRetryImage */,
                                   preImageOpTime);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {incomingStmtId});
        ASSERT_FALSE(foundOps[0].getPreImageOpTime());

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteHasPostImage) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    auto postImageOplogEntry =
        makeNoopOp(BSON("_id" << 1 << "postImage" << true), lsid, incomingTxnNumber, {{1, 0}, 1});
    {
        auto opCtx = makeOperationContext();
        insertOp(opCtx.get(), postImageOplogEntry.getEntry().toBSON());
    }
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1),
                                   lsid,
                                   incomingTxnNumber,
                                   {incomingStmtId},
                                   boost::none /* needsRetryImage */,
                                   boost::none /* preImageOpTime */,
                                   postImageOplogEntry.getOpTime() /* postImageOpTime */);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 2U);
        ASSERT_BSONOBJ_BINARY_EQ(foundOps[0].getObject(), BSON("_id" << 1 << "postImage" << true));
        checkGeneratedNoop(foundOps[1], lsid, incomingTxnNumber, {incomingStmtId});
        ASSERT_EQ(foundOps[1].getPostImageOpTime(), foundOps[0].getOpTime()) << foundOps[1];

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[1]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
    }
}

DEATH_TEST_REGEX_F(ReshardingOplogSessionApplicationTest,
                   IncomingRetryableWriteHasPostImageThatCannotBeFound,
                   "Tripwire assertion.*6344401") {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    repl::OpTime postImageOpTime{{1, 0}, 1};
    auto oplogEntry = makeUpdateOp(BSON("_id" << 1),
                                   lsid,
                                   incomingTxnNumber,
                                   {incomingStmtId},
                                   boost::none /* needsRetryImage */,
                                   boost::none /* preImageOpTime */,
                                   postImageOpTime);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {incomingStmtId});
        ASSERT_FALSE(foundOps[0].getPostImageOpTime());

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
    }
}

// Resharding converts oplog with retry image to old style no-op oplog pairs in normal cases. But
// if it was not able to extract the document from the image collection, it will return the oplog
// entry as is. This is how resharding oplog application can encounter oplog with needsRetryImage.
TEST_F(ReshardingOplogSessionApplicationTest, IncomingRetryableWriteHasNeedsRetryImage) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;
    StmtId incomingStmtId = 2;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    auto oplogEntry = makeUpdateOp(BSON("_id" << 1),
                                   lsid,
                                   incomingTxnNumber,
                                   {incomingStmtId},
                                   repl::RetryImageEnum::kPreImage);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {incomingStmtId});
        ASSERT_FALSE(foundOps[0].getPostImageOpTime());
        ASSERT_FALSE(foundOps[0].getPreImageOpTime());
        ASSERT_FALSE(foundOps[0].getNeedsRetryImage());

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }

    {
        auto opCtx = makeOperationContext();
        checkStatementExecuted(opCtx.get(), lsid, incomingTxnNumber, incomingStmtId);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingTxnForNewSession) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber incomingTxnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    auto oplogEntry = makeFinishTxnOp(lsid, incomingTxnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {kIncompleteHistoryStmtId});

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingTxnHasHigherTxnNumber) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber existingTxnNumber = 100;
    StmtId existingStmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, existingTxnNumber, {existingStmtId});
    }();

    TxnNumber incomingTxnNumber = existingTxnNumber + 1;
    auto oplogEntry = makeFinishTxnOp(lsid, incomingTxnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, incomingTxnNumber, {kIncompleteHistoryStmtId});

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingTxnHasLowerTxnNumber) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber existingTxnNumber = 100;
    StmtId existingStmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, existingTxnNumber, {existingStmtId});
    }();

    TxnNumber incomingTxnNumber = existingTxnNumber - 1;
    auto oplogEntry = makeFinishTxnOp(lsid, incomingTxnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        ASSERT_EQ(sessionTxnRecord->getSessionId(), lsid) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getTxnNum(), existingTxnNumber) << sessionTxnRecord->toBSON();
        ASSERT_EQ(sessionTxnRecord->getLastWriteOpTime(), opTime) << sessionTxnRecord->toBSON();
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingTxnHasEqualTxnNumberAsRetryableWrite) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber txnNumber = 100;
    StmtId existingStmtId = 3;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), lsid, txnNumber, {existingStmtId});
    }();

    auto oplogEntry = makeFinishTxnOp(lsid, txnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 1U);
        checkGeneratedNoop(foundOps[0], lsid, txnNumber, {kIncompleteHistoryStmtId});
        ASSERT_EQ(foundOps[0].getPrevWriteOpTimeInTransaction(), opTime) << foundOps[0];

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_TRUE(bool(sessionTxnRecord));
        checkSessionTxnRecord(*sessionTxnRecord, foundOps[0]);
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingTxnHasEqualTxnNumberAsTxn) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber txnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        auto opTime = makePreparedTxn(opCtx.get(), lsid, txnNumber);
        abortTxn(opCtx.get(), lsid, txnNumber);
        return opTime;
    }();

    auto oplogEntry = makeFinishTxnOp(lsid, txnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IncomingTxnHasHigherTxnNumberThanPreparedTxn) {
    auto lsid = makeLogicalSessionIdForTest();

    TxnNumber existingTxnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return makePreparedTxn(opCtx.get(), lsid, existingTxnNumber);
    }();

    TxnNumber incomingTxnNumber = existingTxnNumber + 1;
    auto oplogEntry = makeFinishTxnOp(lsid, incomingTxnNumber);

    auto conflictingTxnCompletionFuture = [&] {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        return applier.tryApplyOperation(opCtx.get(), oplogEntry);
    }();

    ASSERT_TRUE(bool(conflictingTxnCompletionFuture));
    ASSERT_FALSE(conflictingTxnCompletionFuture->isReady());

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }

    {
        auto opCtx = makeOperationContext();
        abortTxn(opCtx.get(), lsid, existingTxnNumber);
    }

    ASSERT_TRUE(conflictingTxnCompletionFuture->isReady());
    ASSERT_OK(conflictingTxnCompletionFuture->getNoThrow());
}

TEST_F(ReshardingOplogSessionApplicationTest, IgnoreIncomingAbortedRetryableInternalTransaction) {
    auto lsid = makeLogicalSessionIdWithTxnNumberAndUUIDForTest();

    TxnNumber incomingTxnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    // 'makeFinishTxnOp' returns an abortTransaction oplog entry.
    auto oplogEntry = makeFinishTxnOp(lsid, incomingTxnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }
}

TEST_F(ReshardingOplogSessionApplicationTest, IgnoreIncomingNonRetryableInternalTransaction) {
    auto lsid = makeLogicalSessionIdWithTxnUUIDForTest();

    TxnNumber incomingTxnNumber = 100;

    auto opTime = [&] {
        auto opCtx = makeOperationContext();
        return insertSessionRecord(opCtx.get(), makeLogicalSessionIdForTest(), 100, {3});
    }();

    auto oplogEntry = makeFinishTxnOp(lsid, incomingTxnNumber);

    {
        auto opCtx = makeOperationContext();
        ReshardingOplogSessionApplication applier{oplogBufferNss()};
        auto conflictingTxnCompletionFuture = applier.tryApplyOperation(opCtx.get(), oplogEntry);
        ASSERT_FALSE(bool(conflictingTxnCompletionFuture));
    }

    {
        auto opCtx = makeOperationContext();
        auto foundOps = findOplogEntriesNewerThan(opCtx.get(), opTime.getTimestamp());
        ASSERT_EQ(foundOps.size(), 0U);

        auto sessionTxnRecord = findSessionRecord(opCtx.get(), lsid);
        ASSERT_FALSE(bool(sessionTxnRecord));
    }
}

}  // namespace
}  // namespace mongo

max_stars_repo_name: benety/mongo

max_issues_repo_name: benety/mongo

max_forks_repo_name: benety/mongo


text:
/**
 * $Id$
 *
 * Copyright (C)
 * 2013 - $Date$
 *     Martin Wolf <boostnumpy@martin-wolf.org>
 *
 * @file boost/numpy/numpy_c_api.hpp
 * @version $Revision$
 * @date $Date$
 * @author Martin Wolf <boostnumpy@martin-wolf.org>
 * @brief This file includes the numpy C API for usage by boost::numpy.
 *
 *        This file is distributed under the Boost Software License,
 *        Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
 *        http://www.boost.org/LICENSE_1_0.txt).
 */
#ifndef BOOST_NUMPY_NUMPY_C_API_HPP_INCLUDED
#define BOOST_NUMPY_NUMPY_C_API_HPP_INCLUDED

#include <Python.h>

// If this header file is only included for declaration purposes the numpy
// python modules do not need to be imported. This should be done only once
// per shared object library. For boost::numpy this is numpy.cpp.
#if !defined(BOOST_NUMPY_INTERNAL_IMPL_MAIN)
    #define NO_IMPORT_ARRAY
//    #define NO_IMPORT_UFUNC
#endif

// Do not use the deprecated old numpy API if there is the new numpy API
// (as of v1.7) available.
#include <numpy/numpyconfig.h>
#ifdef NPY_1_7_API_VERSION
    #define NPY_NO_DEPRECATED_API NPY_1_7_API_VERSION
#endif

#include <numpy/arrayobject.h>
// Note: The numpy/__ufunc_api.h header file included by numpy/ufuncobject.h
//       has probably a bug: The "static int _import_umath(void)" function is
//       not put into an "#if !defined(NO_IMPORT_UFUNC) && !defined(NO_IMPORT)"
//       condition statement causing an unused function compiler warning when
//       including numpy/ufuncobject.h into more than one internal source file.
//
//       Since we do not need it yet, we just do not include it at all.
//#include <numpy/ufuncobject.h>

// Include a header file that defines the new numpy constants as of numpy
// version 1.7.
#if NPY_FEATURE_VERSION < 0x00000007
#include <boost/numpy/post_v1_7_constants.hpp>
#endif

#endif // !BOOST_NUMPY_NUMPY_C_API_HPP_INCLUDED

ext: hpp

max_stars_repo_name: IceCube-SPNO/BoostNumpy

max_issues_repo_name: IceCube-SPNO/BoostNumpy

max_forks_repo_name: IceCube-SPNO/BoostNumpy


text:
/////////1/////////2/////////3/////////4/////////5/////////6/////////7/////////8
// test_optional.cpp

// (C) Copyright 2004 Pavel Vozenilek
// Use, modification and distribution is subject to the Boost Software
// License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)

// should pass compilation and execution

#include <cstddef> // NULL
#include <cstdio> // remove
#include <fstream>

#include <boost/config.hpp>

#if defined(BOOST_NO_STDC_NAMESPACE)
namespace std{
    using ::remove;
}
#endif

#include <boost/archive/archive_exception.hpp>
#include "test_tools.hpp"

#include <boost/serialization/optional.hpp>
#include <boost/serialization/string.hpp>

struct A {
    int m_x;
    template<class Archive>
    void serialize(Archive & ar, const unsigned int /* version */){
        ar & boost::serialization::make_nvp("x", m_x);
    };
    bool operator==(const A & rhs) const {
        return m_x == rhs.m_x;
    }
    // note that default constructor is not trivial
    A() :
        m_x(0)
    {}
    A(int x) :
        m_x(x)
    {}
};

int test_main( int /* argc */, char* /* argv */[] )
{
    const char * testfile = boost::archive::tmpnam(NULL);
    BOOST_REQUIRE(NULL != testfile);

    const boost::optional<int> aoptional1;
    const boost::optional<int> aoptional2(123);
    const boost::optional<A> aoptional3;
    A a(1);
    const boost::optional<A> aoptional4(a);
    const boost::optional<A *> aoptional5;
    const boost::optional<A *> aoptional6(& a);
    {
        test_ostream os(testfile, TEST_STREAM_FLAGS);
        test_oarchive oa(os, TEST_ARCHIVE_FLAGS);
        oa << boost::serialization::make_nvp("aoptional1",aoptional1);
        oa << boost::serialization::make_nvp("aoptional2",aoptional2);
        oa << boost::serialization::make_nvp("aoptional3",aoptional3);
        oa << boost::serialization::make_nvp("aoptional4",aoptional4);
        oa << boost::serialization::make_nvp("aoptional5",aoptional5);
        oa << boost::serialization::make_nvp("aoptional6",aoptional6);
    }
    boost::optional<int> aoptional1a(999);
    boost::optional<int> aoptional2a;
    boost::optional<A> aoptional3a;
    boost::optional<A> aoptional4a;
    boost::optional<A *> aoptional5a;
    boost::optional<A *> aoptional6a;
    {
        test_istream is(testfile, TEST_STREAM_FLAGS);
        test_iarchive ia(is, TEST_ARCHIVE_FLAGS);
        ia >> boost::serialization::make_nvp("aoptional1",aoptional1a);
        ia >> boost::serialization::make_nvp("aoptional2",aoptional2a);
        ia >> boost::serialization::make_nvp("aoptional3",aoptional3a);
        ia >> boost::serialization::make_nvp("aoptional4",aoptional4a);
        ia >> boost::serialization::make_nvp("aoptional5",aoptional5a);
        ia >> boost::serialization::make_nvp("aoptional6",aoptional6a);
    }
    BOOST_CHECK(aoptional1 == aoptional1a);
    BOOST_CHECK(aoptional2 == aoptional2a);
    BOOST_CHECK(aoptional3 == aoptional3a);
    BOOST_CHECK(aoptional4.get() == aoptional4a.get());
    BOOST_CHECK(aoptional5 == aoptional5a);
    BOOST_CHECK(*aoptional6.get() == *aoptional6a.get());

    std::remove(testfile);
    return EXIT_SUCCESS;
}

// EOF


text:
/*
 * This file belongs to the Galois project, a C++ library for exploiting parallelism.
 * The code is being released under the terms of the 3-Clause BSD License (a
 * copy is located in LICENSE.txt at the top-level directory).
 *
 * Copyright (C) 2018, The University of Texas at Austin. All rights reserved.
 * UNIVERSITY EXPRESSLY DISCLAIMS ANY AND ALL WARRANTIES CONCERNING THIS
 * SOFTWARE AND DOCUMENTATION, INCLUDING ANY WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR ANY PARTICULAR PURPOSE, NON-INFRINGEMENT AND WARRANTIES OF
 * PERFORMANCE, AND ANY WARRANTY THAT MIGHT OTHERWISE ARISE FROM COURSE OF
 * DEALING OR USAGE OF TRADE.  NO WARRANTY IS EITHER EXPRESS OR IMPLIED WITH
 * RESPECT TO THE USE OF THE SOFTWARE OR DOCUMENTATION. Under no circumstances
 * shall University be liable for incidental, special, indirect, direct or
 * consequential damages or loss of profits, interruption of business, or
 * related expenses which may arise from use of Software or Documentation,
 * including but not limited to those resulting from defects in Software and/or
 * Documentation, or loss or inaccuracy of data of any kind.
 */

#include <iostream>
#include <limits>
#include <cmath>
#include "galois/DistGalois.h"
#include "galois/gstl.h"
#include "DistBenchStart.h"

#include "galois/DReducible.h"
#include "galois/AtomicWrapper.h"
#include "galois/ArrayWrapper.h"
#include "galois/runtime/Tracer.h"

#include "galois/graphs/DistributedGraphLoader.h"
#include <boost/serialization/vector.hpp>
#include <boost/serialization/array.hpp>
#include <boost/archive/binary_oarchive.hpp>
#include <boost/archive/binary_iarchive.hpp>
using namespace boost::archive;
// For resilience
#include "resilience.h"

#ifdef __GALOIS_HET_CUDA__
#include "galois/cuda/cuda_device.h"
#include "gen_cuda.h"
struct CUDA_Context* cuda_ctx;
#endif

constexpr static const char* const regionname = "SGD";

/******************************************************************************/
/* Declaration of command line arguments */
/******************************************************************************/

namespace cll = llvm::cl;

static cll::opt<unsigned int>
    maxIterations("maxIterations",
                  cll::desc("Maximum iterations: Default 10000"),
                  cll::init(10000));
static cll::opt<bool> bipartite(
    "bipartite",
    cll::desc("Is graph bipartite? if yes, it expects first N nodes to have "
              "edges."),
    cll::init(false));
static cll::opt<double>
    LEARNING_RATE("LEARNING_RATE",
                  cll::desc("Learning rate (GAMMA): Default 0.00001"),
                  cll::init(0.00001));
static cll::opt<double> LAMBDA("LAMBDA", cll::desc("LAMBDA: Default 0.0001"),
                               cll::init(0.0001));
static cll::opt<double>
    DECAY_RATE("DECAY_RATE",
               cll::desc("Decay rate to be used in step size function "
                         "(DECAY_RATE): Default 0.9"),
               cll::init(0.9));
static cll::opt<double> tolerance(
    "tolerance",
    cll::desc("rms normalized tolerance for convergence:Default 0.01"),
    cll::init(0.01));

/******************************************************************************/
/* Graph structure declarations + helper functions + other initialization */
/******************************************************************************/

#define LATENT_VECTOR_SIZE 20
// static const double LEARNING_RATE = 0.00001; // GAMMA, Purdue: 0.01 Intel:
// 0.001 static const double DECAY_RATE = 0.9; // STEP_DEC, Purdue: 0.1 Intel:
// 0.9 static const double LAMBDA = 0.0001; // Purdue: 1.0 Intel: 0.001
static const double MINVAL = -1e+100;
static const double MAXVAL = 1e+100;

const unsigned int infinity = std::numeric_limits<unsigned int>::max() / 4;

struct NodeData {

  // galois::CopyableArray<galois::CopyableAtomic<double>, LATENT_VECTOR_SIZE>
  // residual_latent_vector; galois::CopyableArray<double, LATENT_VECTOR_SIZE>
  // latent_vector;

  std::vector<double> latent_vector;
  std::vector<galois::CopyableAtomic<double>> residual_latent_vector;

  template <class Archive>
  void serialize(Archive& ar, const unsigned int version) {
    ar& boost::serialization::make_array(latent_vector.data(),
                                         LATENT_VECTOR_SIZE);
    // ar & boost::serialization::make_array(residual_latent_vector.data(),
    // LATENT_VECTOR_SIZE);
  }
};

galois::DynamicBitSet bitset_latent_vector;
galois::DynamicBitSet bitset_residual_latent_vector;

typedef galois::graphs::DistGraph<NodeData, double> Graph;
// typedef galois::graphs::DistGraph<NodeData, uint32_t> Graph;
typedef typename Graph::GraphNode GNode;

#include "gen_sync.hh"
// TODO: Set seed
static double genRand() {
  // generate a random double in (-1,1)
  return 2.0 * ((double)std::rand() / (double)RAND_MAX) - 1.0;
}

static double genVal(uint32_t n) {
  return 2.0 * ((double)n / (double)RAND_MAX) - 1.0;
}

// Purdue learning function
double getstep_size(unsigned int round) {
  return LEARNING_RATE * 1.5 / (1.0 + DECAY_RATE * pow(round + 1, 1.5));
}

/**
 * Prediction of edge weight based on 2 latent vectors
 */
double calcPrediction(const NodeData& movie_data, const NodeData& user_data) {
  double pred = galois::innerProduct(movie_data.latent_vector,
                                     user_data.latent_vector, 0.0);
  double p    = pred;

  pred = std::min(MAXVAL, pred);
  pred = std::max(MINVAL, pred);

#ifndef NDEBUG
  if (p != pred)
    std::cerr << "clamped " << p << " to " << pred << "\n";
#endif

  return pred;
}

/******************************************************************************/
/* Algorithm structures */
/******************************************************************************/

struct InitializeGraph {
  Graph* graph;

  InitializeGraph(Graph* _graph) : graph(_graph) {}

  void static go(Graph& _graph) {
    auto& allNodes = _graph.allNodesRange();

#ifdef __GALOIS_HET_CUDA__
    if (personality == GPU_CUDA) {
      std::string impl_str(_graph.get_run_identifier("InitializeGraph"));
      galois::StatTimer StatTimer_cuda(impl_str.c_str());
      StatTimer_cuda.start();
      InitializeGraph_cuda(*allNodes.begin(), *allNodes.end(), cuda_ctx);
      StatTimer_cuda.stop();
    } else if (personality == CPU)
#endif
      galois::do_all(galois::iterate(allNodes.begin(), allNodes.end()),
                     InitializeGraph{&_graph},
                     galois::loopname("InitializeGraph"));

    // due to latent_vector being generated randomly, it should be sync'd
    // to 1 consistent version across all hosts
    _graph.sync<writeSource, readAny, Reduce_set_latent_vector,
                Broadcast_latent_vector>("InitializeGraph");
  }

  void operator()(GNode src) const {
    NodeData& sdata = graph->getData(src);

    // resize vectors
    sdata.latent_vector.resize(LATENT_VECTOR_SIZE);
    sdata.residual_latent_vector.resize(LATENT_VECTOR_SIZE);

    for (int i = 0; i < LATENT_VECTOR_SIZE; i++) {
      sdata.latent_vector[i] = genVal(src); // randomly create latent vector
      sdata.residual_latent_vector[i] = 0;  // randomly create latent vector

#ifndef NDEBUG
      if (!std::isnormal(sdata.latent_vector[i]))
        galois::gDebug("GEN for ", i, " ", sdata.latent_vector[i]);
#endif
    }
  }
};

struct setMasterBitset {
  Graph* graph;

  setMasterBitset(Graph* _graph) : graph(_graph) {}

  void static go(Graph& _graph) {
    galois::do_all(galois::iterate(_graph.masterNodesRange().begin(),
                                   _graph.masterNodesRange().end()),
                   setMasterBitset{&_graph},
                   galois::loopname("InitializeGraph_crashed_setMasterBiset"));
  }

  void operator()(GNode src) const { bitset_latent_vector.set(src); }
};

struct InitializeGraph_crashed {
  Graph* graph;

  InitializeGraph_crashed(Graph* _graph) : graph(_graph) {}

  void static go(Graph& _graph) {

    setMasterBitset::go(_graph);

    auto& allNodes = _graph.allNodesRange();

#ifdef __GALOIS_HET_CUDA__
    if (personality == GPU_CUDA) {
      std::string impl_str(_graph.get_run_identifier("InitializeGraph"));
      galois::StatTimer StatTimer_cuda(impl_str.c_str());
      StatTimer_cuda.start();
      InitializeGraph_cuda(*allNodes.begin(), *allNodes.end(), cuda_ctx);
      StatTimer_cuda.stop();
    } else if (personality == CPU)
#endif
      galois::do_all(galois::iterate(allNodes.begin(), allNodes.end()),
                     InitializeGraph_crashed{&_graph},
                     galois::loopname("InitializeGraph_crashed"));

    _graph.sync<writeAny, readAny, Reduce_set_latent_vector,
                Broadcast_latent_vector, Bitset_latent_vector>(
        "InitializeGraph_crashed");
  }

  void operator()(GNode src) const {
    NodeData& sdata = graph->getData(src);

    // resize vectors
    sdata.latent_vector.resize(LATENT_VECTOR_SIZE);
    sdata.residual_latent_vector.resize(LATENT_VECTOR_SIZE);

    for (int i = 0; i < LATENT_VECTOR_SIZE; i++) {
      sdata.latent_vector[i] = genVal(src); // randomly create latent vector
      sdata.residual_latent_vector[i] = 0;  // randomly create latent vector

#ifndef NDEBUG
      if (!std::isnormal(sdata.latent_vector[i]))
        galois::gDebug("GEN for ", i, " ", sdata.latent_vector[i]);
#endif
    }
  }
};

struct InitializeGraph_healthy {
  Graph* graph;

  InitializeGraph_healthy(Graph* _graph) : graph(_graph) {}

  void static go(Graph& _graph) {
    auto& allNodes = _graph.allNodesRange();

#ifdef __GALOIS_HET_CUDA__
    if (personality == GPU_CUDA) {
      std::string impl_str(_graph.get_run_identifier("InitializeGraph"));
      galois::StatTimer StatTimer_cuda(impl_str.c_str());
      StatTimer_cuda.start();
      InitializeGraph_cuda(*allNodes.begin(), *allNodes.end(), cuda_ctx);
      StatTimer_cuda.stop();
    } else if (personality == CPU)
#endif
      galois::do_all(galois::iterate(allNodes.begin(), allNodes.end()),
                     InitializeGraph_healthy{&_graph},
                     galois::loopname("InitializeGraph_healthy"));

    _graph.sync<writeAny, readAny, Reduce_set_latent_vector,
                Broadcast_latent_vector, Bitset_latent_vector>(
        "InitializeGraph_healthy");
  }

  void operator()(GNode src) const {
    NodeData& sdata = graph->getData(src);

    for (int i = 0; i < LATENT_VECTOR_SIZE; i++) {
      sdata.residual_latent_vector[i] = 0; // randomly create latent vector
    }
    bitset_latent_vector.set(src);
  }
};

/* Recovery to be called by resilience based fault tolerance
 * It is a NoOp
 */
struct recovery {
  Graph* graph;

  recovery(Graph* _graph) : graph(_graph) {}

  void static go(Graph& _graph) {}
};

struct SGD_mergeResidual {
  Graph* graph;

  SGD_mergeResidual(Graph* _graph) : graph(_graph) {}

  void static go(Graph& _graph) {

    auto& allNodes = _graph.allNodesRange();

#ifdef __GALOIS_HET_CUDA__
    if (personality == GPU_CUDA) {
      std::string impl_str("SGD_" + (_graph.get_run_identifier()));
      galois::StatTimer StatTimer_cuda(impl_str.c_str());
      StatTimer_cuda.start();
      int __retval = 0;
      SGD_all_cuda(__retval, cuda_ctx);
      // DGAccumulator_accum += __retval;
      StatTimer_cuda.stop();
    } else if (personality == CPU)
#endif

      galois::do_all(
          galois::iterate(allNodes.begin(), allNodes.end()),
          SGD_mergeResidual{&_graph},
          galois::loopname(_graph.get_run_identifier("SGD_merge").c_str()),
          galois::steal(), galois::no_stats());
  }

  void operator()(GNode src) const {
    NodeData& sdata              = graph->getData(src);
    auto& latent_vector          = sdata.latent_vector;
    auto& residual_latent_vector = sdata.residual_latent_vector;

    for (int i = 0; i < LATENT_VECTOR_SIZE; ++i) {
      latent_vector[i] += residual_latent_vector[i];
      residual_latent_vector[i] = 0;

#ifndef NDEBUG
      if (!std::isnormal(sdata.latent_vector[i]))
        galois::gDebug("GEN for ", i, " ", sdata.latent_vector[i]);
#endif
    }
  }
};

struct SGD {
  Graph* graph;
  double step_size;
  galois::DGAccumulator<double>& DGAccumulator_accum;

  SGD(Graph* _graph, double _step_size, galois::DGAccumulator<double>& _dga)
      : graph(_graph), step_size(_step_size), DGAccumulator_accum(_dga) {}

  void static go(Graph& _graph, galois::DGAccumulator<double>& dga) {
    unsigned _num_iterations              = 0;
    unsigned _num_iterations_stepSize     = 0;
    unsigned _num_iterations_checkpointed = 0;
    double rms_normalized                 = 0.0;
    double last                           = -1.0;
    double last_checkpointed              = -1.0;
    auto& nodesWithEdges                  = _graph.allNodesWithEdgesRange();
    do {
      // Checkpointing the all the node data
      if (enableFT && recoveryScheme == CP) {
        saveCheckpointToDisk(_num_iterations, _graph);
        // Saving other state variables
        if (_num_iterations % checkpointInterval == 0) {
          last_checkpointed            = last;
          _num_iterations_checkpointed = _num_iterations;
        }
      }

      auto step_size = getstep_size(_num_iterations_stepSize);
      dga.reset();
      galois::do_all(galois::iterate(nodesWithEdges),
                     SGD(&_graph, step_size, dga),
                     galois::loopname(_graph.get_run_identifier("SGD").c_str()),
                     galois::steal(), galois::no_stats());

      _graph.sync<writeDestination, readAny,
                  Reduce_pair_wise_add_array_residual_latent_vector,
                  Broadcast_residual_latent_vector,
                  Bitset_residual_latent_vector>("SGD");

      SGD_mergeResidual::go(_graph);

      /**************************CRASH SITE : start
       * *****************************************/
      if (enableFT && (_num_iterations == crashIteration)) {
        crashSite<recovery, InitializeGraph_crashed, InitializeGraph_healthy>(
            _graph);
        ++_num_iterations;
        if (recoveryScheme == CP) {
          _num_iterations_stepSize = _num_iterations_checkpointed;
          last                     = last_checkpointed;
        }

        continue;
      }
      /**************************CRASH SITE : end
       * *****************************************/

      // calculate root mean squared error
      // Divide by 2 since for symmetric graph it is counted twice
      double error   = dga.reduce() / 2;
      rms_normalized = std::sqrt(error / _graph.globalSizeEdges());

      double error_change = std::abs((last - error) / last);
      if (galois::runtime::getSystemNetworkInterface().ID == 0) {
        galois::gPrint("ITERATION : ", _num_iterations, "\n");
        galois::gDebug("RMS Normalized : ", rms_normalized);
        galois::gPrint("RMS : ", rms_normalized, "\n");
        galois::gPrint("abs(last - error/last) : ", error_change, "\n");
      }

      if (error_change < tolerance) {
        break;
      }
      last = error;
      ++_num_iterations;
      ++_num_iterations_stepSize;
    } while ((_num_iterations < maxIterations));

    if (galois::runtime::getSystemNetworkInterface().ID == 0) {
      galois::runtime::reportStat_Single(
          regionname, "NumIterations_" + std::to_string(_graph.get_run_num()),
          (unsigned long)_num_iterations);
    }
  }

  void operator()(GNode src) const {
    NodeData& sdata           = graph->getData(src);
    auto& movie_node          = sdata.latent_vector;
    auto& residual_movie_node = sdata.residual_latent_vector;

    for (auto jj = graph->edge_begin(src), ej = graph->edge_end(src); jj != ej;
         ++jj) {
      GNode dst   = graph->getEdgeDst(jj);
      auto& ddata = graph->getData(dst);

      auto& user_node          = ddata.latent_vector;
      auto& residual_user_node = ddata.residual_latent_vector;
      // auto& sdata_up = sdata.updates;

      double edge_rating = graph->getEdgeData(jj);

      // doGradientUpdate
      double old_dp = galois::innerProduct(user_node, movie_node, double(0));

      double cur_error = edge_rating - old_dp;
      DGAccumulator_accum += (cur_error * cur_error);

      assert(cur_error < 10000 && cur_error > -10000);

      bool setBit = false;
      // update both vectors based on error derived from 2 previous vectors
      for (int i = 0; i < LATENT_VECTOR_SIZE; ++i) {

        double prevUser  = user_node[i];
        double prevMovie = movie_node[i];

        // Only update the destination
        galois::atomicAdd(
            residual_user_node[i],
            double(step_size * (cur_error * prevMovie - LAMBDA * prevUser)));
        // galois::gPrint("val : ", residual_user_node[i], "\n");
        assert(std::isnormal(residual_user_node[i]));
        if (!setBit && std::abs(residual_user_node[i]) > 0.1)
          setBit = true;

        // galois::atomicAdd(residual_movie_node[i],  double(step_size *
        // (cur_error * prevUser - LAMBDA * prevMovie)));
        // assert(std::isnormal(residual_movie_node[i]));
      }
      if (setBit)
        bitset_residual_latent_vector.set(dst);
    }
  }
};

/******************************************************************************/
/* Main */
/******************************************************************************/
constexpr static const char* const name = "SGD - Distributed Heterogeneous";
constexpr static const char* const desc = "SGD on Distributed Galois.";
constexpr static const char* const url  = 0;

int main(int argc, char** argv) {
  galois::DistMemSys G;
  DistBenchStart(argc, argv, name, desc, url);

  const auto& net = galois::runtime::getSystemNetworkInterface();
  if (net.ID == 0) {
    galois::runtime::reportParam(regionname, "Max Iterations",
                                 (unsigned long)maxIterations);

    galois::runtime::reportParam(regionname, "ENABLE_FT", (enableFT));
  }

  galois::StatTimer StatTimer_total("TimerTotal", regionname);

  StatTimer_total.start();
#ifdef __GALOIS_HET_CUDA__
  Graph* hg = distGraphInitialization<NodeData, double>(&cuda_ctx);
#else
  Graph* hg = distGraphInitialization<NodeData, double>();
#endif

  // bitset comm setup
  bitset_latent_vector.resize(hg->size());
  bitset_residual_latent_vector.resize(hg->size());

  galois::gPrint("[", net.ID, "] InitializeGraph::go called\n");

  galois::StatTimer StatTimer_init("TIMER_GRAPH_INIT", regionname);
  StatTimer_init.start();
  InitializeGraph::go((*hg));
  StatTimer_init.stop();

  galois::runtime::getHostBarrier().wait();

  // accumulators for use in operators
  galois::DGAccumulator<double> DGAccumulator_accum;
  // galois::DGAccumulator<uint64_t> DGAccumulator_sum;
  // galois::DGAccumulator<uint32_t> DGAccumulator_max;
  // galois::GReduceMax<uint32_t> m;

  for (auto run = 0; run < numRuns; ++run) {
    galois::gPrint("[", net.ID, "] SGD::go run ", run, " called\n");
    std::string timer_str("Timer_" + std::to_string(run));
    galois::StatTimer StatTimer_main(timer_str.c_str(), regionname);

    StatTimer_main.start();
    SGD::go((*hg), DGAccumulator_accum);
    StatTimer_main.stop();

    if ((run + 1) != numRuns) {
#ifdef __GALOIS_HET_CUDA__
      if (personality == GPU_CUDA) {
        // bitset_dist_current_reset_cuda(cuda_ctx);
      } else
#endif
        bitset_latent_vector.reset();
      bitset_residual_latent_vector.reset();

      (*hg).set_num_run(run + 1);
      InitializeGraph::go((*hg));
      galois::runtime::getHostBarrier().wait();
    }
  }

  StatTimer_total.stop();

  return 0;
}

max_stars_repo_name: vancemiller/Galois

max_issues_repo_name: vancemiller/Galois

max_forks_repo_name: vancemiller/Galois


text:
// Copyright (C) 2020 T. Zachary Laine
//
// Distributed under the Boost Software License, Version 1.0. (See
// accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
#ifndef BOOST_TEXT_DETAIL_ROPE_HPP
#define BOOST_TEXT_DETAIL_ROPE_HPP

#include <boost/text/string_view.hpp>
#include <boost/text/transcode_iterator.hpp>

#include <boost/text/detail/btree.hpp>

namespace boost { namespace text { namespace detail {

    constexpr std::size_t string_insert_max = BOOST_TEXT_STRING_INSERT_MAX;

    static_assert(sizeof(node_ptr<char, std::string>) * 8 <= 64, "");

    struct segment_inserter
    {
        template<typename Iter, typename Sentinel>
        void operator()(Iter first, Sentinel last) const
        {
            for (; os_.good() && first != last; ++first) {
                os_ << *first;
            }
        }

        std::ostream & os_;
    };

}}}

#endif