text:
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 <atomic>
#include <cstdlib>
#include <functional>
#include <memory>
#include <ostream>
#include <string>
#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>

#include <boost/optional/optional.hpp>
#include <glog/logging.h>
#include <gtest/gtest.h>

#include "kudu/client/client.h"
#include "kudu/client/schema.h"
#include "kudu/client/shared_ptr.h"
#include "kudu/common/common.pb.h"
#include "kudu/common/table_util.h"
#include "kudu/common/wire_protocol.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/hms/hms_client.h"
#include "kudu/hms/mini_hms.h"
#include "kudu/integration-tests/cluster_itest_util.h"
#include "kudu/integration-tests/external_mini_cluster-itest-base.h"
#include "kudu/integration-tests/hms_itest-base.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/master.proxy.h"
#include "kudu/master/sentry_authz_provider-test-base.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/rpc_header.pb.h"
#include "kudu/rpc/user_credentials.h"
#include "kudu/security/test/mini_kdc.h"
#include "kudu/sentry/mini_sentry.h"
#include "kudu/sentry/sentry_client.h"
#include "kudu/sentry/sentry_policy_service_types.h"
#include "kudu/thrift/client.h"
#include "kudu/util/monotime.h"
#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/slice.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"

using boost::make_optional;
using boost::none;
using boost::optional;
using kudu::client::KuduClient;
using kudu::client::KuduColumnSchema;
using kudu::client::KuduScanToken;
using kudu::client::KuduScanTokenBuilder;
using kudu::client::KuduSchema;
using kudu::client::KuduTable;
using kudu::client::KuduTableAlterer;
using kudu::client::sp::shared_ptr;
using kudu::hms::HmsClient;
using kudu::master::AlterRoleGrantPrivilege;
using kudu::master::CreateRoleAndAddToGroups;
using kudu::master::GetDatabasePrivilege;
using kudu::master::GetTableLocationsResponsePB;
using kudu::master::GetTabletLocationsResponsePB;
using kudu::master::GetTablePrivilege;
using kudu::master::MasterServiceProxy;
using kudu::master::ResetAuthzCacheRequestPB;
using kudu::master::ResetAuthzCacheResponsePB;
using kudu::master::VOTER_REPLICA;
using kudu::rpc::RpcController;
using kudu::rpc::UserCredentials;
using kudu::sentry::SentryClient;
using sentry::TSentryGrantOption;
using sentry::TSentryPrivilege;
using std::atomic;
using std::function;
using std::ostream;
using std::string;
using std::thread;
using std::unique_ptr;
using std::unordered_set;
using std::vector;
using strings::Substitute;

namespace kudu {

// Parameters for the operator functor (see below for OperationFunc).
struct OperationParams {
  // NOLINTNEXTLINE(google-explicit-constructor)
  OperationParams(string table_name,
                  string new_table_name = "")
      : table_name(std::move(table_name)),
        new_table_name(std::move(new_table_name)) {
  }
  const string table_name;
  const string new_table_name;
};

// Parameters for the privilege functor (see below for PrivilegeFunc).
struct PrivilegeParams {
  // NOLINTNEXTLINE(google-explicit-constructor)
  PrivilegeParams(string db_name,
                  string table_name = "")
      : db_name(std::move(db_name)),
        table_name(std::move(table_name)) {
  }
  const string db_name;
  const string table_name;
};

// Test Master authorization enforcement with Sentry and HMS
// integration enabled.
class SentryITestBase : public HmsITestBase {
 public:
  static const char* const kAdminGroup;
  static const char* const kAdminUser;
  static const char* const kUserGroup;
  static const char* const kTestUser;
  static const char* const kImpalaUser;
  static const char* const kDevRole;
  static const char* const kAdminRole;
  static const char* const kDatabaseName;
  static const char* const kTableName;
  static const char* const kSecondTable;

  Status StopSentry() {
    RETURN_NOT_OK(sentry_client_->Stop());
    RETURN_NOT_OK(cluster_->sentry()->Stop());
    return Status::OK();
  }

  Status StartSentry() {
    RETURN_NOT_OK(cluster_->sentry()->Start());
    RETURN_NOT_OK(cluster_->kdc()->Kinit("kudu"));
    RETURN_NOT_OK(sentry_client_->Start());
    return Status::OK();
  }

  Status GetTableLocationsWithTableId(const string& table_name,
                                      optional<const string&> table_id) {
    const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
    std::shared_ptr<MasterServiceProxy> proxy = cluster_->master_proxy();
    UserCredentials user_credentials;
    user_credentials.set_real_user(kTestUser);
    proxy->set_user_credentials(user_credentials);

    GetTableLocationsResponsePB table_locations;
    return itest::GetTableLocations(proxy, table_name, kTimeout, VOTER_REPLICA,
                                    table_id, &table_locations);
  }

  Status GrantCreateTablePrivilege(const PrivilegeParams& p) {
    return AlterRoleGrantPrivilege(sentry_client_.get(), kDevRole,
        GetDatabasePrivilege(p.db_name, "CREATE"));
  }

  Status GrantDropTablePrivilege(const PrivilegeParams& p) {
    return AlterRoleGrantPrivilege(sentry_client_.get(), kDevRole,
        GetTablePrivilege(p.db_name, p.table_name, "DROP"));
  }

  Status GrantAlterTablePrivilege(const PrivilegeParams& p) {
    return AlterRoleGrantPrivilege(sentry_client_.get(), kDevRole,
        GetTablePrivilege(p.db_name, p.table_name, "ALTER"));
  }

  Status GrantRenameTablePrivilege(const PrivilegeParams& p) {
    RETURN_NOT_OK(AlterRoleGrantPrivilege(sentry_client_.get(), kDevRole,
        GetTablePrivilege(p.db_name, p.table_name, "ALL")));
    return AlterRoleGrantPrivilege(sentry_client_.get(), kDevRole,
        GetDatabasePrivilege(p.db_name, "CREATE"));
  }

  Status GrantGetMetadataTablePrivilege(const PrivilegeParams& p) {
    return AlterRoleGrantPrivilege(sentry_client_.get(), kDevRole,
        GetTablePrivilege(p.db_name, p.table_name, "METADATA"));
  }

  Status CreateTable(const OperationParams& p) {
    Slice hms_database;
    Slice hms_table;
    RETURN_NOT_OK(ParseHiveTableIdentifier(p.table_name,
                                           &hms_database, &hms_table));
    return CreateKuduTable(hms_database.ToString(), hms_table.ToString());
  }

  Status IsCreateTableDone(const OperationParams& p) {
    bool in_progress = false;
    return client_->IsCreateTableInProgress(p.table_name, &in_progress);
  }

  Status DropTable(const OperationParams& p) {
    return client_->DeleteTable(p.table_name);
  }

  Status AlterTable(const OperationParams& p) {
    unique_ptr<KuduTableAlterer> alterer(
        client_->NewTableAlterer(p.table_name));
    return alterer->DropColumn("int32_val")->Alter();
  }

  Status IsAlterTableDone(const OperationParams& p) {
    bool in_progress = false;
    return client_->IsAlterTableInProgress(p.table_name, &in_progress);
  }

  Status RenameTable(const OperationParams& p) {
    unique_ptr<KuduTableAlterer> alterer(
        client_->NewTableAlterer(p.table_name));
    return alterer->RenameTo(p.new_table_name)->Alter();
  }

  Status GetTableSchema(const OperationParams& p) {
    KuduSchema schema;
    return client_->GetTableSchema(p.table_name, &schema);
  }

  Status GetTableLocations(const OperationParams& p) {
    return GetTableLocationsWithTableId(p.table_name, /*table_id=*/none);
  }

  Status GetTabletLocations(const OperationParams& p) {
    // Log in as 'test-admin' to get the tablet ID.
    RETURN_NOT_OK(cluster_->kdc()->Kinit(kAdminUser));
    shared_ptr<KuduClient> client;
    RETURN_NOT_OK(cluster_->CreateClient(nullptr, &client));

    shared_ptr<KuduTable> kudu_table;
    RETURN_NOT_OK(client->OpenTable(p.table_name, &kudu_table));
    vector<KuduScanToken*> tokens;
    ElementDeleter deleter(&tokens);
    KuduScanTokenBuilder builder(kudu_table.get());
    RETURN_NOT_OK(builder.Build(&tokens));
    CHECK(!tokens.empty());
    const string& tablet_id = tokens[0]->tablet().id();

    // Log back as 'test-user'.
    RETURN_NOT_OK(cluster_->kdc()->Kinit(kTestUser));

    static const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
    std::shared_ptr<MasterServiceProxy> proxy(cluster_->master_proxy());
    UserCredentials user_credentials;
    user_credentials.set_real_user(kTestUser);
    proxy->set_user_credentials(user_credentials);

    GetTabletLocationsResponsePB tablet_locations;
    return itest::GetTabletLocations(proxy, tablet_id, kTimeout,
                                     VOTER_REPLICA, &tablet_locations);
  }

  void SetUp() override {
    HmsITestBase::SetUp();

    cluster::ExternalMiniClusterOptions opts;
    // Always enable Kerberos, as sentry deployment does not make much sense
    // in a non-Kerberized environment.
    bool enable_kerberos = true;
    opts.hms_mode = HmsMode::ENABLE_METASTORE_INTEGRATION;
    opts.enable_sentry = true;
    opts.enable_kerberos = enable_kerberos;
    // Configure the timeout to reduce the run time of tests that involve
    // re-connections.
    const string timeout = AllowSlowTests() ? "5" : "2";
    opts.extra_master_flags.emplace_back(
       Substitute("$0=$1", "--sentry_service_send_timeout_seconds", timeout));
    opts.extra_master_flags.emplace_back(
       Substitute("$0=$1", "--sentry_service_recv_timeout_seconds", timeout));

    // Add 'impala' as trusted user who may access the cluster without being
    // authorized.
    opts.extra_master_flags.emplace_back("--trusted_user_acl=impala");
    opts.extra_master_flags.emplace_back("--user_acl=test-user,impala");

    // Enable/disable caching of authz privileges received from Sentry.
    opts.extra_master_flags.emplace_back(
        Substitute("--sentry_privileges_cache_capacity_mb=$0",
                   IsAuthzPrivilegeCacheEnabled() ? 1 : 0));

    StartClusterWithOpts(std::move(opts));

    // Create principals 'impala' and 'kudu'. Configure to use the latter.
    ASSERT_OK(cluster_->kdc()->CreateUserPrincipal(kImpalaUser));
    ASSERT_OK(cluster_->kdc()->CreateUserPrincipal("kudu"));
    ASSERT_OK(cluster_->kdc()->Kinit("kudu"));

    thrift::ClientOptions hms_opts;
    hms_opts.enable_kerberos = enable_kerberos;
    hms_opts.service_principal = "hive";
    hms_client_.reset(new HmsClient(cluster_->hms()->address(), hms_opts));
    ASSERT_OK(hms_client_->Start());

    thrift::ClientOptions sentry_opts;
    sentry_opts.enable_kerberos = enable_kerberos;
    sentry_opts.service_principal = "sentry";
    sentry_client_.reset(new SentryClient(cluster_->sentry()->address(), sentry_opts));
    ASSERT_OK(sentry_client_->Start());

    // User to Role mapping:
    // 1. user -> developer,
    // 2. admin -> admin.
    ASSERT_OK(CreateRoleAndAddToGroups(sentry_client_.get(), kDevRole, kUserGroup));
    ASSERT_OK(CreateRoleAndAddToGroups(sentry_client_.get(), kAdminRole, kAdminGroup));

    // Grant privilege 'ALL' on database 'db' to role admin.
    TSentryPrivilege privilege = GetDatabasePrivilege(
        kDatabaseName, "ALL",
        TSentryGrantOption::DISABLED);
    ASSERT_OK(AlterRoleGrantPrivilege(sentry_client_.get(),
                                              kAdminRole, privilege));

    // Create database 'db' in HMS and Kudu tables 'table' and 'second_table' which
    // owns by user 'test-admin'.
    ASSERT_OK(CreateDatabase(kDatabaseName));
    ASSERT_OK(CreateKuduTable(kDatabaseName, kTableName));
    ASSERT_OK(CreateKuduTable(kDatabaseName, kSecondTable));
    NO_FATALS(CheckTable(kDatabaseName, kTableName,
                         make_optional<const string&>(kAdminUser)));
    NO_FATALS(CheckTable(kDatabaseName, kSecondTable,
                         make_optional<const string&>(kAdminUser)));

    // Log in as 'test-user' and reset the client to pick up the change in user.
    ASSERT_OK(cluster_->kdc()->Kinit(kTestUser));
    ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
  }

  void TearDown() override {
    if (sentry_client_) {
      ASSERT_OK(sentry_client_->Stop());
    }
    if (hms_client_) {
      ASSERT_OK(hms_client_->Stop());
    }
    HmsITestBase::TearDown();
  }

  virtual bool IsAuthzPrivilegeCacheEnabled() const {
    return false;
  }

 protected:
  unique_ptr<SentryClient> sentry_client_;
};

const char* const SentryITestBase::kAdminGroup = "admin";
const char* const SentryITestBase::kAdminUser = "test-admin";
const char* const SentryITestBase::kUserGroup = "user";
const char* const SentryITestBase::kTestUser = "test-user";
const char* const SentryITestBase::kImpalaUser = "impala";
const char* const SentryITestBase::kDevRole = "developer";
const char* const SentryITestBase::kAdminRole = "ad";
const char* const SentryITestBase::kDatabaseName = "db";
const char* const SentryITestBase::kTableName = "table";
const char* const SentryITestBase::kSecondTable = "second_table";

// Functor that performs a certain operation (e.g. Create, Rename) on a table
// given its name and its desired new name, if necessary (only used for Rename).
typedef function<Status(SentryITestBase*, const OperationParams&)> OperatorFunc;

// Functor that grants the required permission for an operation (e.g Create,
// Rename) on a table given the database the table belongs to and the name of
// the table, if applicable.
typedef function<Status(SentryITestBase*, const PrivilegeParams&)> PrivilegeFunc;

// A description of the operation function that describes a particular action
// on a table a user can perform, as well as the privilege granting function
// that grants the required permission to the user to perform the action.
struct AuthzFuncs {
  OperatorFunc do_action;
  PrivilegeFunc grant_privileges;
  string description;
};
ostream& operator <<(ostream& out, const AuthzFuncs& d) {
  out << d.description;
  return out;
}

// A description of an authorization process, including the protected resource (table),
// the operation function, as well as the privilege granting function.
struct AuthzDescriptor {
  AuthzFuncs funcs;
  string database;
  string table_name;
  string new_table_name;
};
ostream& operator <<(ostream& out, const AuthzDescriptor& d) {
  out << d.funcs.description;
  return out;
}

// Test basic master authorization enforcement with Sentry and HMS integration
// enabled.
class MasterSentryTest : public SentryITestBase {};

// Test that the trusted user can access the cluster without being authorized.
TEST_F(MasterSentryTest, TestTrustedUserAcl) {
  // Log in as 'impala' and reset the client to pick up the change in user.
  ASSERT_OK(cluster_->kdc()->Kinit(kImpalaUser));
  ASSERT_OK(cluster_->CreateClient(nullptr, &client_));

  vector<string> tables;
  ASSERT_OK(client_->ListTables(&tables));
  unordered_set<string> tables_set(tables.begin(), tables.end());
  ASSERT_EQ(unordered_set<string>({ Substitute("$0.$1", kDatabaseName, kTableName),
                                    Substitute("$0.$1", kDatabaseName, kSecondTable) }),
            tables_set);

  ASSERT_OK(CreateKuduTable(kDatabaseName, "new_table"));
  NO_FATALS(CheckTable(kDatabaseName, "new_table",
                       make_optional<const string&>(kImpalaUser)));
}

TEST_F(MasterSentryTest, TestAuthzListTables) {
  // ListTables is not parameterized as other operations below, because non-authorized
  // tables will be filtered instead of returning NOT_AUTHORIZED error.
  const auto table_name = Substitute("$0.$1", kDatabaseName, kTableName);
  const auto sec_table_name = Substitute("$0.$1", kDatabaseName, kSecondTable);

  // Listing tables shows nothing without proper privileges.
  vector<string> tables;
  ASSERT_OK(client_->ListTables(&tables));
  ASSERT_TRUE(tables.empty());

  // Listing tables only shows the tables which user has proper privileges on.
  tables.clear();
  ASSERT_OK(GrantGetMetadataTablePrivilege({ kDatabaseName, kTableName }));
  ASSERT_OK(client_->ListTables(&tables));
  ASSERT_EQ(vector<string>({ table_name }), tables);

  tables.clear();
  ASSERT_OK(GrantGetMetadataTablePrivilege({ kDatabaseName, kSecondTable }));
  ASSERT_OK(client_->ListTables(&tables));
  unordered_set<string> tables_set(tables.begin(), tables.end());
  ASSERT_EQ(unordered_set<string>({ table_name, sec_table_name }), tables_set);
}

// When authorizing ListTables, if there is a concurrent rename, we may not end
// up showing the table.
TEST_F(MasterSentryTest, TestAuthzListTablesConcurrentRename) {
  ASSERT_OK(cluster_->SetFlag(cluster_->master(),
      "catalog_manager_inject_latency_list_authz_ms", "3000"));;
  const auto table_name = Substitute("$0.$1", kDatabaseName, kTableName);
  const auto sec_table_name = Substitute("$0.$1", kDatabaseName, kSecondTable);
  ASSERT_OK(GrantGetMetadataTablePrivilege({ kDatabaseName, kTableName }));
  ASSERT_OK(GrantRenameTablePrivilege({ kDatabaseName, kTableName }));

  // List the tables while injecting latency.
  vector<string> tables;
  thread t([&] {
    ASSERT_OK(client_->ListTables(&tables));
  });

  // While that's happening, rename one of the tables.
  ASSERT_OK(RenameTable({ table_name, Substitute("$0.$1", kDatabaseName, "b") }));
  NO_FATALS(t.join());

  // We shouldn't see the renamed table.
  ASSERT_EQ(1, tables.size());
  ASSERT_EQ(sec_table_name, tables[0]);
}

TEST_F(MasterSentryTest, TestTableOwnership) {
  ASSERT_OK(GrantCreateTablePrivilege({ kDatabaseName }));
  ASSERT_OK(CreateKuduTable(kDatabaseName, "new_table"));
  NO_FATALS(CheckTable(kDatabaseName, "new_table",
                       make_optional<const string&>(kTestUser)));

  // Checks the user with table ownership automatically has ALL privilege on the
  // table. User 'test-user' can delete the same table without specifically granting
  // 'DROP ON TABLE'. Note that ownership population between the HMS and the Sentry
  // service happens synchronously, therefore, the table deletion should succeed
  // right after the table creation.
  // TODO(hao): test create a table with a different owner than the client’s username?
  ASSERT_OK(client_->DeleteTable(Substitute("$0.$1", kDatabaseName, "new_table")));
  NO_FATALS(CheckTableDoesNotExist(kDatabaseName, "new_table"));
}

// Checks Sentry privileges are synchronized upon table rename in the HMS.
TEST_F(MasterSentryTest, TestRenameTablePrivilegeTransfer) {
  ASSERT_OK(GrantRenameTablePrivilege({ kDatabaseName, kTableName }));
  ASSERT_OK(RenameTable({ Substitute("$0.$1", kDatabaseName, kTableName),
                          Substitute("$0.$1", kDatabaseName, "b") }));
  NO_FATALS(CheckTable(kDatabaseName, "b",
                       make_optional<const string&>(kAdminUser)));

  unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(
      Substitute("$0.$1", kDatabaseName, "b")));
  alterer->DropColumn("int16_val");

  // Note that unlike table creation, there could be a delay between the table renaming
  // in Kudu and the privilege renaming in Sentry. Because Kudu uses the transactional
  // listener of the HMS to get notification of table alteration events, while Sentry
  // uses post event listener (which is executed outside the HMS transaction). There
  // is a chance that Kudu already finish the table renaming but the privilege renaming
  // hasn't been reflected in the Sentry service.
  ASSERT_EVENTUALLY([&] {
    ASSERT_OK(alterer->Alter());
  });
  NO_FATALS(CheckTable(kDatabaseName, "b", make_optional<const string&>(kAdminUser)));
}

class TestAuthzTable :
    public MasterSentryTest,
    public ::testing::WithParamInterface<AuthzDescriptor> {
    // TODO(aserbin): update the test to introduce authz privilege caching
};

TEST_P(TestAuthzTable, TestAuthorizeTable) {
  const AuthzDescriptor& desc = GetParam();
  const auto table_name = Substitute("$0.$1", desc.database, desc.table_name);
  const auto new_table_name = Substitute("$0.$1",
                                         desc.database, desc.new_table_name);
  const OperationParams action_params = { table_name, new_table_name };
  const PrivilegeParams privilege_params = { desc.database, desc.table_name };

  // User 'test-user' attempts to operate on the table without proper privileges.
  Status s = desc.funcs.do_action(this, action_params);
  ASSERT_TRUE(s.IsNotAuthorized()) << s.ToString();
  ASSERT_STR_CONTAINS(s.ToString(), "unauthorized action");

  // User 'test-user' can operate on the table after obtaining proper privileges.
  ASSERT_OK(desc.funcs.grant_privileges(this, privilege_params));
  ASSERT_OK(desc.funcs.do_action(this, action_params));

  // Ensure that operating on a table while the Sentry is unreachable fails.
  ASSERT_OK(StopSentry());
  s = desc.funcs.do_action(this, action_params);
  ASSERT_TRUE(s.IsNetworkError()) << s.ToString();
}

static const AuthzDescriptor kAuthzCombinations[] = {
    {
      {
        &SentryITestBase::CreateTable,
        &SentryITestBase::GrantCreateTablePrivilege,
        "CreateTable",
      },
      SentryITestBase::kDatabaseName,
      "new_table",
      ""
    },
    {
      {
        &SentryITestBase::DropTable,
        &SentryITestBase::GrantDropTablePrivilege,
        "DropTable",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
    {
      {
        &SentryITestBase::AlterTable,
        &SentryITestBase::GrantAlterTablePrivilege,
        "AlterTable",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
    {
      {
        &SentryITestBase::RenameTable,
        &SentryITestBase::GrantRenameTablePrivilege,
        "RenameTable",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      "new_table"
    },
    {
      {
        &SentryITestBase::GetTableSchema,
        &SentryITestBase::GrantGetMetadataTablePrivilege,
        "GetTableSchema",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
    {
      {
        &SentryITestBase::GetTableLocations,
        &SentryITestBase::GrantGetMetadataTablePrivilege,
        "GetTableLocations",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
    {
      {
        &SentryITestBase::GetTabletLocations,
        &SentryITestBase::GrantGetMetadataTablePrivilege,
        "GetTabletLocations",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
    {
      {
        &SentryITestBase::IsCreateTableDone,
        &SentryITestBase::GrantGetMetadataTablePrivilege,
        "IsCreateTableDone",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
    {
      {
        &SentryITestBase::IsAlterTableDone,
        &SentryITestBase::GrantGetMetadataTablePrivilege,
        "IsAlterTableDone",
      },
      SentryITestBase::kDatabaseName,
      SentryITestBase::kTableName,
      ""
    },
};
INSTANTIATE_TEST_CASE_P(AuthzCombinations,
                        TestAuthzTable,
                        ::testing::ValuesIn(kAuthzCombinations));

// Test that when the client passes a table identifier with the table name
// and table ID refer to different tables, the client needs permission on
// both tables for returning TABLE_NOT_FOUND error to avoid leaking table
// existence.
TEST_F(MasterSentryTest, TestMismatchedTable) {
  const auto table_name_a = Substitute("$0.$1", kDatabaseName, kTableName);
  const auto table_name_b = Substitute("$0.$1", kDatabaseName, kSecondTable);

  // Log in as 'test-admin' to get the tablet ID.
  ASSERT_OK(cluster_->kdc()->Kinit(kAdminUser));
  shared_ptr<KuduClient> client;
  ASSERT_OK(cluster_->CreateClient(nullptr, &client));
  shared_ptr<KuduTable> table;
  ASSERT_OK(client->OpenTable(table_name_a, &table));
  optional<const string&> table_id_a = make_optional<const string&>(table->id());

  // Log back as 'test-user'.
  ASSERT_OK(cluster_->kdc()->Kinit(kTestUser));

  Status s = GetTableLocationsWithTableId(table_name_b, table_id_a);
  ASSERT_TRUE(s.IsNotAuthorized());
  ASSERT_STR_CONTAINS(s.ToString(), "unauthorized action");

  ASSERT_OK(GrantGetMetadataTablePrivilege({ kDatabaseName, kTableName }));
  s = GetTableLocationsWithTableId(table_name_b, table_id_a);
  ASSERT_TRUE(s.IsNotAuthorized());
  ASSERT_STR_CONTAINS(s.ToString(), "unauthorized action");

  ASSERT_OK(GrantGetMetadataTablePrivilege({ kDatabaseName, kSecondTable }));
  s = GetTableLocationsWithTableId(table_name_b, table_id_a);
  ASSERT_TRUE(s.IsNotFound());
  ASSERT_STR_CONTAINS(s.ToString(), "the table ID refers to a different table");
}

class AuthzErrorHandlingTest :
    public SentryITestBase,
    public ::testing::WithParamInterface<AuthzFuncs> {
    // TODO(aserbin): update the test to introduce authz privilege caching
};
TEST_P(AuthzErrorHandlingTest, TestNonExistentTable) {
  static constexpr const char* const kTableName = "non_existent";
  const AuthzFuncs& funcs = GetParam();
  const auto table_name = Substitute("$0.$1", kDatabaseName, kTableName);
  const auto new_table_name = Substitute("$0.$1", kDatabaseName, "b");
  const OperationParams action_params = { table_name, new_table_name };
  const PrivilegeParams privilege_params = { kDatabaseName, kTableName };

  // Ensure that operating on a non-existent table without proper privileges gives
  // a NOT_AUTHORIZED error, instead of leaking the table existence by giving a
  // TABLE_NOT_FOUND error.
  Status s = funcs.do_action(this, action_params);
  ASSERT_TRUE(s.IsNotAuthorized());
  ASSERT_STR_CONTAINS(s.ToString(), "unauthorized action");

  // Ensure that operating on a non-existent table fails
  // while Sentry is unreachable.
  ASSERT_OK(StopSentry());
  s = funcs.do_action(this, action_params);
  ASSERT_TRUE(s.IsNetworkError()) << s.ToString();

  // Ensure that operating on a non-existent table with proper privileges gives a
  // TABLE_NOT_FOUND error.
  ASSERT_OK(StartSentry());
  ASSERT_EVENTUALLY([&] {
    // SentryAuthzProvider throttles reconnections, so it's necessary
    // to wait out the backoff.
    ASSERT_OK(funcs.grant_privileges(this, privilege_params));
  });
  s = funcs.do_action(this, action_params);
  ASSERT_TRUE(s.IsNotFound()) << s.ToString();
}

static const AuthzFuncs kAuthzFuncCombinations[] = {
    {
      &SentryITestBase::DropTable,
      &SentryITestBase::GrantDropTablePrivilege,
      "DropTable"
    },
    {
      &SentryITestBase::AlterTable,
      &SentryITestBase::GrantAlterTablePrivilege,
      "AlterTable"
    },
    {
      &SentryITestBase::RenameTable,
      &SentryITestBase::GrantRenameTablePrivilege,
      "RenameTable"
    },
    {
      &SentryITestBase::GetTableSchema,
      &SentryITestBase::GrantGetMetadataTablePrivilege,
      "GetTableSchema"
    },
    {
      &SentryITestBase::GetTableLocations,
      &SentryITestBase::GrantGetMetadataTablePrivilege,
      "GetTableLocations"
    },
    {
      &SentryITestBase::IsCreateTableDone,
      &SentryITestBase::GrantGetMetadataTablePrivilege,
      "IsCreateTableDone"
    },
    {
      &SentryITestBase::IsAlterTableDone,
      &SentryITestBase::GrantGetMetadataTablePrivilege,
      "IsAlterTableDone"
    },
};

INSTANTIATE_TEST_CASE_P(AuthzFuncCombinations,
                        AuthzErrorHandlingTest,
                        ::testing::ValuesIn(kAuthzFuncCombinations));

// Class for test scenarios verifying functionality of managing AuthzProvider's
// privileges cache via Kudu RPC.
class SentryAuthzProviderCacheITest : public SentryITestBase {
 public:
  bool IsAuthzPrivilegeCacheEnabled() const override {
    return true;
  }

  Status ResetCache() {
    // ResetAuthzCache() RPC requires admin/superuser credentials, so this
    // method calls Kinit(kAdminUser) to authenticate appropriately. However,
    // it's necessary to return back the credentials of the regular user after
    // resetting the cache since the rest of the scenario is supposed to run
    // without superuser credentials.
    SCOPED_CLEANUP({
      WARN_NOT_OK(cluster_->kdc()->Kinit(kTestUser),
                  "could not restore Kerberos credentials");
    });
    RETURN_NOT_OK(cluster_->kdc()->Kinit(kAdminUser));
    std::shared_ptr<MasterServiceProxy> proxy = cluster_->master_proxy();
    UserCredentials user_credentials;
    user_credentials.set_real_user(kAdminUser);
    proxy->set_user_credentials(user_credentials);

    RpcController ctl;
    ResetAuthzCacheResponsePB res;
    RETURN_NOT_OK(proxy->ResetAuthzCache(
        ResetAuthzCacheRequestPB(), &res, &ctl));
    return res.has_error() ? StatusFromPB(res.error().status()) : Status::OK();
  }
};

// This test scenario uses AlterTable() to make sure AuthzProvider's cache
// empties upon successful ResetAuthzCache() RPC.
TEST_F(SentryAuthzProviderCacheITest, AlterTable) {
  const auto table_name = Substitute("$0.$1", kDatabaseName, kTableName);
  ASSERT_OK(GrantAlterTablePrivilege({ kDatabaseName, kTableName }));
  {
    unique_ptr<KuduTableAlterer> table_alterer(
        client_->NewTableAlterer(table_name)->DropColumn("int8_val"));
    auto s = table_alterer->Alter();
    ASSERT_TRUE(s.ok()) << s.ToString();
  }
  ASSERT_OK(StopSentry());
  {
    unique_ptr<KuduTableAlterer> table_alterer(
        client_->NewTableAlterer(table_name)->DropColumn("int16_val"));
    auto s = table_alterer->Alter();
    ASSERT_TRUE(s.ok()) << s.ToString();
  }
  ASSERT_OK(ResetCache());
  {
    // After resetting the cache, it should not be possible to perform another
    // ALTER TABLE operation: no entries are in the cache, so authz provider
    // needs to fetch information from Sentry directly.
    unique_ptr<KuduTableAlterer> table_alterer(
        client_->NewTableAlterer(table_name)->DropColumn("int32_val"));
    auto s = table_alterer->Alter();
    ASSERT_TRUE(s.IsNetworkError()) << s.ToString();
  }

  // Try to do the same after starting Sentry back. It should be a success.
  ASSERT_OK(StartSentry());
  {
    unique_ptr<KuduTableAlterer> table_alterer(
        client_->NewTableAlterer(table_name)->DropColumn("int32_val"));
    auto s = table_alterer->Alter();
    ASSERT_TRUE(s.ok()) << s.ToString();
  }
}

// This test scenario calls a couple of authz methods of SentryAuthzProvider
// while resetting its fetcher's cache in parallel using master's
// ResetAuthzCache() RPC.
TEST_F(SentryAuthzProviderCacheITest, ResetAuthzCacheConcurrentAlterTable) {
  constexpr const auto num_threads = 16;
  const auto run_interval = AllowSlowTests() ? MonoDelta::FromSeconds(8)
                                             : MonoDelta::FromSeconds(1);
  ASSERT_OK(GrantCreateTablePrivilege({ kDatabaseName }));
  for (auto idx = 0; idx < num_threads; ++idx) {
    ASSERT_OK(CreateTable(Substitute("$0.$1", kDatabaseName, idx)));
    ASSERT_OK(GrantAlterTablePrivilege({ kDatabaseName, Substitute("$0", idx) }));
  }

  vector<Status> threads_task_status(num_threads);
  {
    atomic<bool> stopped(false);
    vector<thread> threads;

    SCOPED_CLEANUP({
      stopped = true;
      for (auto& thread : threads) {
        thread.join();
      }
    });

    for (auto idx = 0; idx < num_threads; ++idx) {
      const auto thread_idx = idx;
      threads.emplace_back([&, thread_idx] () {
        const auto table_name = Substitute("$0.$1", kDatabaseName, thread_idx);

        while (!stopped) {
          SleepFor(MonoDelta::FromMicroseconds((rand() % 2 + 1) * thread_idx));
          {
            unique_ptr<KuduTableAlterer> table_alterer(
               client_->NewTableAlterer(table_name));
            table_alterer->DropColumn("int8_val");

            auto s = table_alterer->Alter();
            if (!s.ok()) {
              threads_task_status[thread_idx] = s;
              return;
            }
          }

          SleepFor(MonoDelta::FromMicroseconds((rand() % 3 + 1) * thread_idx));
          {
            unique_ptr<KuduTableAlterer> table_alterer(
                client_->NewTableAlterer(table_name));
            table_alterer->AddColumn("int8_val")->Type(KuduColumnSchema::INT8);
            auto s = table_alterer->Alter();
            if (!s.ok()) {
              threads_task_status[thread_idx] = s;
              return;
            }
          }
        }
      });
    }

    const auto time_beg = MonoTime::Now();
    const auto time_end = time_beg + run_interval;
    while (MonoTime::Now() < time_end) {
      SleepFor(MonoDelta::FromMilliseconds((rand() % 3)));
      ASSERT_OK(ResetCache());
    }
  }
  for (auto idx = 0; idx < threads_task_status.size(); ++idx) {
    SCOPED_TRACE(Substitute("results for thread $0", idx));
    const auto& s = threads_task_status[idx];
    EXPECT_TRUE(s.ok()) << s.ToString();
  }
}

// This test scenario documents an artifact of the Kudu+HMS+Sentry integration
// when authz cache is enabled (the cache is enabled by default). In essence,
// information on the ownership of a table created during a short period of
// Sentry's unavailability will not ever appear in Sentry. That might be
// misleading because CreateTable() reports a success to the client. The created
// table indeed exists and is fully functional otherwise, but the corresponding
// owner privilege record is absent in Sentry.
//
// TODO(aserbin): clarify why it works with HEAD of the master branches
//                of Sentry/Hive but fails with Sentry 2.1.0 and Hive 2.1.1.
TEST_F(SentryAuthzProviderCacheITest, DISABLED_CreateTables) {
  constexpr const char* const kGhostTables[] = { "t10", "t11" };

  // Grant CREATE TABLE and METADATA privileges on the database.
  ASSERT_OK(GrantCreateTablePrivilege({ kDatabaseName }));
  ASSERT_OK(AlterRoleGrantPrivilege(
      sentry_client_.get(), kDevRole,
      GetDatabasePrivilege(kDatabaseName, "METADATA")));

  // Make sure it's possible to create a table in the database. This also
  // populates the privileges cache with information on the privileges
  // granted on the database.
  ASSERT_OK(CreateKuduTable(kDatabaseName, "t0"));

  // An attempt to open a not-yet-existing table will fetch the information
  // on the granted privileges on the table into the privileges cache.
  for (const auto& t : kGhostTables) {
    shared_ptr<KuduTable> kudu_table;
    const auto s = client_->OpenTable(Substitute("$0.$1", kDatabaseName, t),
                                      &kudu_table);
    ASSERT_TRUE(s.IsNotFound()) << s.ToString();
  }

  ASSERT_OK(StopSentry());

  // CreateTable() with operation timeout longer than HMS --> Sentry
  // communication timeout successfully completes. After failing to push
  // the information on the newly created table to Sentry due to the logic
  // implemented in the SentrySyncHMSNotificationsPostEventListener plugin,
  // HMS sends success response to Kudu master and Kudu successfully completes
  // the rest of the steps.
  ASSERT_OK(CreateKuduTable(kDatabaseName, kGhostTables[0]));

  // In this case, the timeout for the CreateTable RPC is set to be lower than
  // the HMS --> Sentry communication timeout (see corresponding parameters
  // of the MiniHms::EnableSentry() method). CreateTable() successfully passes
  // the authz phase since the information on privileges is cached and no
  // Sentry RPC calls are attempted. However, since Sentry is down,
  // CreateTable() takes a long time on the HMS's side and the client's
  // request times out, while the creation of the table continues in the
  // background.
  {
    const auto s = CreateKuduTable(kDatabaseName, kGhostTables[1],
                                   MonoDelta::FromSeconds(1));
    ASSERT_TRUE(s.IsTimedOut()) << s.ToString();
  }

  // Before starting Sentry, make sure the abandoned request to create the
  // latter table succeeded even if CreateKuduTable() reported timeout.
  // This is to make sure HMS stopped trying to push notification
  // on table creation to Sentry anymore via the metastore plugin
  // SentrySyncHMSNotificationsPostEventListener.
  ASSERT_EVENTUALLY([&]{
    bool exists;
    ASSERT_OK(client_->TableExists(
        Substitute("$0.$1", kDatabaseName, kGhostTables[1]), &exists));
    ASSERT_TRUE(exists);
  });

  ASSERT_OK(ResetCache());

  // After resetting the cache, it should not be possible to create another
  // table: authz provider needs to fetch information on privileges directly
  // from Sentry, but it's still down.
  {
    const auto s = CreateKuduTable(kDatabaseName, "t2");
    ASSERT_TRUE(s.IsNetworkError()) << s.ToString();
  }

  ASSERT_OK(StartSentry());

  // Try to create the table after starting Sentry back: it should be a success.
  ASSERT_OK(CreateKuduTable(kDatabaseName, "t2"));

  // Once table has been created, it should be possible to perform DDL operation
  // on it since the user is the owner of the table.
  {
    unique_ptr<KuduTableAlterer> table_alterer(
        client_->NewTableAlterer(Substitute("$0.$1", kDatabaseName, "t2")));
    table_alterer->AddColumn("new_int8_columun")->Type(KuduColumnSchema::INT8);
    ASSERT_OK(table_alterer->Alter());
  }

  // Try to run DDL against the tables created during Sentry's downtime. These
  // should not be authorized since Sentry didn't received information on the
  // ownership of those tables from HMS during their creation and there isn't
  // any catch up for those events made after Sentry started.
  for (const auto& t : kGhostTables) {
    unique_ptr<KuduTableAlterer> table_alterer(
        client_->NewTableAlterer(Substitute("$0.$1", kDatabaseName, t)));
    table_alterer->AddColumn("new_int8_columun")->Type(KuduColumnSchema::INT8);
    auto s = table_alterer->Alter();
    ASSERT_TRUE(s.IsNotAuthorized()) << s.ToString();

    // After granting the ALTER TABLE privilege the alteration should be
    // successful. One caveat: it's necessary to reset the cache since the cache
    // will not have the new record till the current entry hasn't yet expired.
    ASSERT_OK(GrantAlterTablePrivilege({ kDatabaseName, t }));
    ASSERT_OK(ResetCache());
    ASSERT_OK(table_alterer->Alter());
  }
}

// Basic test to verify access control and functionality of
// the ResetAuthzCache(); integration with Sentry is not enabled.
class AuthzCacheControlTest : public ExternalMiniClusterITestBase {
 public:
  void SetUp() override {
    ExternalMiniClusterITestBase::SetUp();
    cluster::ExternalMiniClusterOptions opts;
    opts.enable_kerberos = true;
    StartClusterWithOpts(opts);
  }

  // Utility method to call master's ResetAuthzCache RPC under the credentials
  // of the specified 'user'. The credentials should have been set appropriately
  // before calling this method (e.g., call kinit if necessary).
  Status ResetCache(const string& user,
                    RpcController* ctl,
                    ResetAuthzCacheResponsePB* resp) {
    RETURN_NOT_OK(cluster_->kdc()->Kinit(user));
    std::shared_ptr<MasterServiceProxy> proxy = cluster_->master_proxy();
    UserCredentials user_credentials;
    user_credentials.set_real_user(user);
    proxy->set_user_credentials(user_credentials);

    ResetAuthzCacheRequestPB req;
    return proxy->ResetAuthzCache(req, resp, ctl);
  }
};

TEST_F(AuthzCacheControlTest, ResetCacheNoSentryIntegration) {
  // Non-admin users (i.e. those not in --superuser_acl) are not allowed
  // to reset authz cache.
  for (const auto& user : { "test-user", "joe-interloper", }) {
    ASSERT_OK(cluster_->kdc()->Kinit(user));
    ResetAuthzCacheResponsePB resp;
    RpcController ctl;
    auto s = ResetCache(user, &ctl, &resp);
    ASSERT_TRUE(s.IsRemoteError()) << s.ToString();
    ASSERT_FALSE(resp.has_error());

    const auto* err_status = ctl.error_response();
    ASSERT_NE(nullptr, err_status);
    ASSERT_TRUE(err_status->has_code());
    ASSERT_EQ(rpc::ErrorStatusPB::ERROR_APPLICATION, err_status->code());
    ASSERT_STR_CONTAINS(err_status->message(),
                        "unauthorized access to method: ResetAuthzCache");
  }

  // The cache can be reset with credentials of a super-user. However, in
  // case if integration with Sentry is not enabled, the AuthzProvider
  // doesn't have any cache.
  {
    ResetAuthzCacheResponsePB resp;
    RpcController ctl;
    const auto s = ResetCache("test-admin", &ctl, &resp);
    ASSERT_TRUE(s.ok()) << s.ToString();
    ASSERT_EQ(nullptr, ctl.error_response());
    ASSERT_TRUE(resp.has_error()) << resp.error().DebugString();
    const auto app_s = StatusFromPB(resp.error().status());
    ASSERT_TRUE(app_s.IsNotSupported()) << app_s.ToString();
    ASSERT_STR_CONTAINS(app_s.ToString(),
                        "provider does not have privileges cache");
  }
}

// A test for the ValidateSentryServiceRpcAddresses group flag validator.
// The only existing test scenario covers only the negative case, while
// several other Sentry-related (and not) tests provide good coverage
// for all the positive cases.
class MasterSentryAndHmsFlagsTest : public KuduTest {
};

TEST_F(MasterSentryAndHmsFlagsTest, MasterRefuseToStart) {
  // The code below results in setting the --sentry_service_rpc_addresses flag
  // to the mini-sentry's RPC address, but leaving the --hive_metastore_uris
  // flag unset (i.e. its value is an empty string). Such a combination of flag
  // settings makes it impossible to start Kudu master.
  cluster::ExternalMiniClusterOptions opts;
  opts.enable_kerberos = true;
  opts.enable_sentry = true;
  opts.hms_mode = HmsMode::NONE;

  cluster::ExternalMiniCluster cluster(std::move(opts));
  const auto s = cluster.Start();
  const auto msg = s.ToString();
  ASSERT_TRUE(s.IsRuntimeError()) << msg;
  ASSERT_STR_CONTAINS(msg, "failed to start masters: Unable to start Master");
  ASSERT_STR_CONTAINS(msg, "kudu-master: process exited with non-zero status 1");
}

} // namespace kudu

ext: cc

max_stars_repo_name: yuyunliuhen/kudu

max_issues_repo_name: TingyLiang/kudu

max_forks_repo_name: TingyLiang/kudu


text:
//
// Copyright (c) 2015, Deutsches Forschungszentrum für Künstliche Intelligenz GmbH.
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
// * Redistributions of source code must retain the above copyright notice, this
//   list of conditions and the following disclaimer.
//
// * Redistributions in binary form must reproduce the above copyright notice,
//   this list of conditions and the following disclaimer in the documentation
//   and/or other materials provided with the distribution.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//

#ifndef envire_octomap_OcTreeVisualization_H
#define envire_octomap_OcTreeVisualization_H

#include <boost/noncopyable.hpp>
#include <vizkit3d/Vizkit3DPlugin.hpp>
#include <osg/Geode>

namespace octomap
{
    class AbstractOcTree;
}

namespace vizkit3d
{
    class OcTreeVisualization
        : public vizkit3d::Vizkit3DPlugin< boost::shared_ptr<octomap::AbstractOcTree> >
        , boost::noncopyable
    {
    Q_OBJECT

    Q_PROPERTY(bool ShowOccupied READ getShowOccupied WRITE setShowOccupied)
    Q_PROPERTY(bool ShowFreespace READ getShowFreespace WRITE setShowFreespace)
    Q_PROPERTY(int MaxTreeDepth READ getTreeMaxDepth WRITE setTreeMaxDepth)
    Q_PROPERTY(double AlphaOccupiedLevel READ getAlphaOccupiedLevel WRITE setAlphaOccupiedLevel)
    Q_PROPERTY(QStringList ColorMode READ getColorModes WRITE setColorModes)

    public:
        OcTreeVisualization();
        ~OcTreeVisualization();

        Q_INVOKABLE void updateData(boost::shared_ptr<octomap::AbstractOcTree> const &sample)
        {vizkit3d::Vizkit3DPlugin< boost::shared_ptr<octomap::AbstractOcTree> >::updateData(sample);}

    public slots:
        bool getShowOccupied() {return show_occupied;}
        void setShowOccupied(bool b) {show_occupied = b; emit propertyChanged("ShowOccupied"); redrawTree();}
        bool getShowFreespace() {return show_freespace;}
        void setShowFreespace(bool b) {show_freespace = b; emit propertyChanged("ShowFreespace"); redrawTree();}
        void setTreeMaxDepth(int max_depth) {this->max_depth = max_depth; emit propertyChanged("MaxTreeDepth"); reloadTree();}
        int getTreeMaxDepth() {return max_depth;}
        bool getAlphaOccupiedLevel() {return alpha_level;}
        void setAlphaOccupiedLevel(double alpha_level) {this->alpha_level = alpha_level; redrawTree();}
        QStringList getColorModes() {return color_modes;}
        void setColorModes(QStringList color_mode) {this->color_modes = color_mode; redrawTree();}

    protected:
        virtual osg::ref_ptr<osg::Node> createMainNode();
        virtual void updateMainNode(osg::Node* node);
        virtual void updateDataIntern(boost::shared_ptr<octomap::AbstractOcTree> const& value);
        void reloadTree() {new_tree = true; setDirty();}
        void redrawTree() {redraw = true; setDirty();}

    private:
        boost::shared_ptr<octomap::AbstractOcTree> tree;
        bool show_occupied;
        bool show_freespace;
        int max_depth;
        double alpha_level;
        bool new_tree;
        bool redraw;
        QStringList color_modes;
        std::map<QString, unsigned> color_mode_map;
    };
}
#endif

ext: hpp


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

#include "validation/impl/stateful_validator_impl.hpp"

#include <string>

#include <boost/algorithm/cxx11/all_of.hpp>
#include <boost/format.hpp>
#include <boost/range/adaptor/filtered.hpp>
#include <boost/range/adaptor/indexed.hpp>
#include <boost/range/adaptor/transformed.hpp>
#include "common/result.hpp"
#include "interfaces/iroha_internal/batch_meta.hpp"
#include "logger/logger.hpp"
#include "validation/utils.hpp"

namespace iroha {
  namespace validation {

    /**
     * Complements initial transaction check with command-by-command check
     * @param temporary_wsv to apply commands on
     * @param transactions_errors_log to write errors to
     * @param tx to be checked
     * @return empty result, if check is successful, command error otherwise
     */
    static bool checkTransactions(
        ametsuchi::TemporaryWsv &temporary_wsv,
        validation::TransactionsErrors &transactions_errors_log,
        const shared_model::interface::Transaction &tx) {
      return temporary_wsv.apply(tx).match(
          [](expected::Value<void> &) { return true; },
          [&tx, &transactions_errors_log](
              expected::Error<validation::CommandError> &error) {
            transactions_errors_log.emplace_back(validation::TransactionError{
                tx.hash(), std::move(error.error)});
            return false;
          });
    };

    /**
     * Validate all transactions supplied; includes special rules, such as batch
     * validation etc
     * @param txs to be validated
     * @param temporary_wsv to apply transactions on
     * @param transactions_errors_log to write errors to
     * @param batch_parser to parse batches from transaction range
     * @return range of transactions, which passed stateful validation
     */
    static auto validateTransactions(
        const shared_model::interface::types::TransactionsCollectionType &txs,
        ametsuchi::TemporaryWsv &temporary_wsv,
        validation::TransactionsErrors &transactions_errors_log,
        const shared_model::interface::TransactionBatchParser &batch_parser) {
      std::vector<bool> validation_results;
      validation_results.reserve(boost::size(txs));

      for (auto batch : batch_parser.parseBatches(txs)) {
        auto validation = [&](auto &tx) {
          return checkTransactions(temporary_wsv, transactions_errors_log, tx);
        };
        if (batch.front().batchMeta()
            and batch.front().batchMeta()->get()->type()
                == shared_model::interface::types::BatchType::ATOMIC) {
          // check all batch's transactions for validness
          auto savepoint = temporary_wsv.createSavepoint(
              "batch_" + batch.front().hash().hex());
          bool validation_result = false;

          if (boost::algorithm::all_of(batch, validation)) {
            // batch is successful; release savepoint
            validation_result = true;
            savepoint->release();
          } else {
            auto failed_tx_hash = transactions_errors_log.back().tx_hash;
            for (const auto &tx : batch) {
              if (tx.hash() != failed_tx_hash) {
                transactions_errors_log.emplace_back(
                    validation::TransactionError{
                        tx.hash(),
                        // TODO igor-egorov 22.01.2019 IR-245 add a separate
                        // error code for failed batch case
                        validation::CommandError{
                            "",
                            1,  // internal error code
                            "Another transaction failed the batch",
                            true,
                            std::numeric_limits<size_t>::max()}});
              }
            }
          }

          validation_results.insert(
              validation_results.end(), boost::size(batch), validation_result);
        } else {
          for (const auto &tx : batch) {
            validation_results.push_back(validation(tx));
          }
        }
      }

      return txs | boost::adaptors::indexed()
          | boost::adaptors::filtered(
                 [validation_results =
                      std::move(validation_results)](const auto &el) {
                   return validation_results.at(el.index());
                 })
          | boost::adaptors::transformed(
                 [](const auto &el) -> decltype(auto) { return el.value(); });
    }

    StatefulValidatorImpl::StatefulValidatorImpl(
        std::unique_ptr<shared_model::interface::UnsafeProposalFactory> factory,
        std::shared_ptr<shared_model::interface::TransactionBatchParser>
            batch_parser,
        logger::LoggerPtr log)
        : factory_(std::move(factory)),
          batch_parser_(std::move(batch_parser)),
          log_(std::move(log)) {}

    std::unique_ptr<validation::VerifiedProposalAndErrors>
    StatefulValidatorImpl::validate(
        const shared_model::interface::Proposal &proposal,
        ametsuchi::TemporaryWsv &temporaryWsv) {
      log_->info("transactions in proposal: {}",
                 proposal.transactions().size());

      auto validation_result = std::make_unique<VerifiedProposalAndErrors>();
      auto valid_txs =
          validateTransactions(proposal.transactions(),
                               temporaryWsv,
                               validation_result->rejected_transactions,
                               *batch_parser_);

      // Since proposal came from ordering gate it was already validated.
      // All transactions are validated as well
      // This allows for unsafe construction of proposal
      validation_result->verified_proposal =
          std::const_pointer_cast<const shared_model::interface::Proposal>(
              std::shared_ptr<shared_model::interface::Proposal>(
                  factory_->unsafeCreateProposal(
                      proposal.height(), proposal.createdTime(), valid_txs)));

      log_->info("transactions in verified proposal: {}",
                 validation_result->verified_proposal->transactions().size());
      return validation_result;
    }
  }  // namespace validation
}  // namespace iroha

ext: cpp

max_stars_repo_name: artyom-yurin/iroha-archive

max_issues_repo_name: artyom-yurin/iroha-archive

max_forks_repo_name: artyom-yurin/iroha-archive


text:
// Copyright (c) 2020 Julian Bernhard, Klemens Esterle, Patrick Hart and
// Tobias Kessler
//
// This work is licensed under the terms of the MIT license.
// For a copy, see <https://opensource.org/licenses/MIT>.

#ifndef BARK_WORLD_MAP_ROAD_CORRIDOR_HPP_
#define BARK_WORLD_MAP_ROAD_CORRIDOR_HPP_

#include <map>
#include <vector>
#include <string>
#include <utility>
#include <boost/functional/hash.hpp>
#include "bark/world/opendrive/opendrive.hpp"
#include "bark/world/map/road.hpp"
#include "bark/world/map/lane.hpp"
#include "bark/world/map/lane_corridor.hpp"
#include "bark/geometry/geometry.hpp"

namespace bark {
namespace world {
namespace map {

using bark::geometry::Line;
using bark::geometry::Polygon;
using bark::geometry::Point2d;
using bark::geometry::Within;
using bark::geometry::BufferPolygon;
using bark::world::opendrive::XodrRoadId;
using bark::world::opendrive::XodrDrivingDirection;

struct RoadCorridor {
  //! Getter
  RoadPtr GetRoad(const RoadId& road_id) const {
    if (roads_.count(road_id) == 0)
      return nullptr;
    return roads_.at(road_id);
  }
  Roads GetRoads() const { return roads_; }
  Polygon GetPolygon() const { return road_polygon_; }
  Lanes GetLanes(const RoadId& road_id) const {
    return this->GetRoad(road_id)->GetLanes();
  }
  LaneCorridorPtr GetLaneCorridor(const LaneId& lane_id) const {
    if (lane_corridors_.count(lane_id) == 0)
      return nullptr;
    return lane_corridors_.at(lane_id);
  }
  std::vector<LaneCorridorPtr> GetUniqueLaneCorridors() const {
    return unique_lane_corridors_;
  }
  std::map<LaneId, LaneCorridorPtr> GetLaneCorridorMap() const {
    return lane_corridors_;
  }
  std::vector<XodrRoadId> GetRoadIds() const {
    return road_ids_;
  }
  LaneCorridorPtr GetCurrentLaneCorridor(const Point2d& pt) const {
    for (const auto& lane_corr : unique_lane_corridors_) {
      if (Collide(pt, lane_corr->GetMergedPolygon()))
        return lane_corr;
    }
    return nullptr;
  }
  LaneCorridorPtr GetNearestLaneCorridor(const Point2d& pt) const;
  std::pair<LaneCorridorPtr, LaneCorridorPtr>
  GetLeftRightLaneCorridor(const Point2d& pt) const;

  static std::size_t GetHash(
    const XodrDrivingDirection& driving_direction,
    const std::vector<XodrRoadId>& road_ids) {
    // calculate hash using driving_direction and road_ids
    std::size_t road_id_hash = boost::hash_range(
      road_ids.begin(),
      road_ids.end());
    boost::hash_combine(road_id_hash, driving_direction);
    return road_id_hash;
  }

  //! Setter
  void SetRoads(const Roads& roads) {
    roads_ = roads;
  }
  void SetLaneCorridor(const LaneId& lane_id,
    const LaneCorridorPtr& corr) {
    lane_corridors_[lane_id] = corr;
    // we only want to push it if it is not included
    if (std::find(unique_lane_corridors_.begin(),
                  unique_lane_corridors_.end(),
                  corr) == unique_lane_corridors_.end()) {
      unique_lane_corridors_.push_back(corr);
    }
  }
  void SetUniqueLaneCorridors(
    const std::vector<LaneCorridorPtr>& unique_lane_corridors) {
    unique_lane_corridors_ = unique_lane_corridors;
  }
  void SetLaneCorridorMap(
    const std::map<LaneId, LaneCorridorPtr>& lane_corridors) {
    lane_corridors_ = lane_corridors;
  }
  bool ComputeRoadPolygon(double buffer_dist = 0.3) {
    Polygon merged_polygon;
    // merge all lane polygons
    for (const auto& lane_corr : unique_lane_corridors_) {
      const auto& lanes = lane_corr->GetLanes();
      for (const auto& lane : lanes) {
        Polygon poly_buffered;
        BufferPolygon(lane.second->GetPolygon(), buffer_dist, &poly_buffered);
        merged_polygon.ConcatenatePolygons(poly_buffered);
      }
    }
    Polygon poly_buffered_merged;
    BufferPolygon(merged_polygon, -buffer_dist, &poly_buffered_merged);
    road_polygon_ = poly_buffered_merged;
    return true;
  }
  void SetPolygon(const Polygon& poly) { road_polygon_ = poly; }
  void SetRoadIds(
    const std::vector<XodrRoadId>& road_ids) {
    road_ids_ = road_ids;
  }

  Roads roads_;
  Polygon road_polygon_;
  std::vector<LaneCorridorPtr> unique_lane_corridors_;
  std::vector<XodrRoadId> road_ids_;
  std::map<LaneId, LaneCorridorPtr> lane_corridors_;
};
using RoadCorridorPtr = std::shared_ptr<RoadCorridor>;

}  // namespace map
}  // namespace world
}  // namespace bark

#endif  // BARK_WORLD_MAP_ROAD_CORRIDOR_HPP_


text:
// Boost.Geometry (aka GGL, Generic Geometry Library)
// Robustness Test
//
// Copyright (c) 2009-2012 Barend Gehrels, Amsterdam, the Netherlands.
// 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)

#ifndef BOOST_GEOMETRY_TEST_ROBUSTNESS_MAKE_SQUARE_POLYGON_HPP
#define BOOST_GEOMETRY_TEST_ROBUSTNESS_MAKE_SQUARE_POLYGON_HPP

#include <boost/geometry.hpp>

template <typename Polygon, typename Generator, typename Settings>
inline void make_square_polygon(Polygon& polygon, Generator& generator, Settings const& settings)
{
    using namespace boost::geometry;

    typedef typename point_type<Polygon>::type point_type;
    typedef typename coordinate_type<Polygon>::type coordinate_type;

    coordinate_type x, y;
    x = generator();
    y = generator();

    typename ring_type<Polygon>::type& ring = exterior_ring(polygon);

    point_type p;
    set<0>(p, x); set<1>(p, y);         append(ring, p);
    set<0>(p, x); set<1>(p, y + 1);     append(ring, p);
    set<0>(p, x + 1); set<1>(p, y + 1); append(ring, p);
    set<0>(p, x + 1); set<1>(p, y);     append(ring, p);
    set<0>(p, x); set<1>(p, y);         append(ring, p);

    if (settings.triangular)
    {
        // Remove a point, generator says which
        int c = generator() % 4;
        if (c >= 1 && c <= 3)
        {
            ring.erase(ring.begin() + c);
        }
    }
}

#endif // BOOST_GEOMETRY_TEST_ROBUSTNESS_MAKE_SQUARE_POLYGON_HPP


text:
#ifndef SKYLARK_COMBBLAS_SLAB_VIEW_HPP
#define SKYLARK_COMBBLAS_SLAB_VIEW_HPP

#if SKYLARK_HAVE_COMBBLAS
#include <CombBLAS.h>
#include <CommGrid.h>
#endif

#include <map>
#include <boost/mpi.hpp>
#include <boost/serialization/map.hpp>
#include <boost/serialization/vector.hpp>

#include "combblas_comm_grid.hpp"

namespace skylark {
namespace utility {

#if SKYLARK_HAVE_COMBBLAS

/**
 * Slab view for CombBLAS matrices.
 * This are helpers for the panel gemms, where we communicate parts of
 * the CombBLAS matrix and then perform a local Elemental gemm.
 * The communication patterns are dictated by the "output" Elemental
 * distribution, e.g. if we want an STAR / MR view, CombBLAS values are
 * redistributed accordingly.
 * Note that due to the CombBLAS distribution we generate slabs in parallel:
 *
 *     [x....|x....]
 *     [x....|x....]
 *     [-----------]
 *     [x....|x....]
 *     [x....|x....]
 *
 * FIXME: - see how we can integrate into base::view
 *        - different comm patterns for _set_view
 *        - compare performance against one-sided value getter for cb matrices
 *        - flag to generate in-order
 *        - transpose case should invert row/col slab view
 */
template <typename index_type, typename value_type>
struct combblas_slab_view_t {

    typedef SpDCCols<index_type, value_type> cb_col_t;
    typedef SpParMat<index_type, value_type, cb_col_t> sp_par_mat_t;
    typedef typename cb_col_t::SpColIter cb_col_itr_t;
    typedef typename cb_col_t::SpColIter::NzIter cb_nz_itr_t;

    /**
     *  Create a slab view of the CombBLAS matrix B that may be transposed.
     */
    combblas_slab_view_t(const sp_par_mat_t &B, bool transp_b = false)
        : _world(B.getcommgrid()->GetWorld(), boost::mpi::comm_duplicate)
        , _data((const_cast<sp_par_mat_t &>(B)).seq())
        , _col_itr(_data.begcol())
        , _row_itr(_data.begnz(_col_itr))
        , _transp_b(transp_b)
        , _offset_row(cb_my_row_offset(B))
        , _offset_col(cb_my_col_offset(B))
        , _global_row(0)
        , _global_col(0) {

        _ncol = B.getncol();
        _nrow = B.getnrow();
        if(_transp_b) {
            _ncol = B.getnrow();
            _nrow = B.getncol();
        }

        for(cb_col_itr_t col = _data.begcol(); col != _data.endcol(); col++)
            _row_itrs.push_back(_data.begnz(col));
    }

    /**
     * Extract the value of a global index from the CombBLAS matrix.
     * TODO: Use this to define a sparse-dense local GEMM that avoids building
     *       a dense part of the CombBLAS matrix.
     */
    value_type operator()(index_type g_row, index_type g_col) {

        index_type map_idx = idx(g_row, g_col);
        if(_values.count(map_idx) > 0)
            return _values[map_idx];
        else
            return static_cast<value_type>(0);
    }

    /**
     *  Extract a Elemental view of CombBLAS COLUMNS. The distribution of the
     *  Elemental view is determined by the distribution of the input matrix
     *  A.
     *  The column_idxs store the global column indices that contain non-zeros.
     */
    template<typename dist_elem_matrix_t>
    void extract_elemental_column_slab_view(
            const dist_elem_matrix_t &A,
            std::set<index_type> &column_idxs,
            size_t width = 1) {

        _values.clear();

        // structure per proc data
        std::vector< std::map<index_type, value_type> >
            per_proc_data(_world.size());

        // accumulate locale values
        for(size_t i = 0; i < width && _col_itr != _data.endcol();
            _col_itr++, i++) {

            // first accumulate column values, then distribute
            for(cb_nz_itr_t nz = _data.begnz(_col_itr);
                nz != _data.endnz(_col_itr); nz++) {

                index_type g_cb_col_idx = _col_itr.colid() + _offset_col;
                index_type g_cb_row_idx = nz.rowid()  + _offset_row;
                if(_transp_b) std::swap(g_cb_row_idx, g_cb_col_idx);

                size_t target_proc = A.Owner(g_cb_row_idx, g_cb_col_idx);
                index_type coords = idx(g_cb_row_idx, g_cb_col_idx);
                per_proc_data[target_proc].insert(
                    std::make_pair(coords, nz.value()));
            }
        }

        std::set<index_type> indices;
        _set_view(per_proc_data, indices);
        typename std::set<index_type>::iterator itr;
        for(itr = indices.begin(); itr != indices.end(); itr++)
            column_idxs.insert(col(*itr));
    }

    /**
     *  Extract a Elemental view of CombBLAS COLUMNS in sequence. The
     *  distribution of the Elemental view is determined by the distribution
     *  of the input matrix A.
     */
    template<typename dist_elem_matrix_t>
    void extract_elemental_column_slab_view(
            const dist_elem_matrix_t &A,
            size_t width = 1) {

        _values.clear();

        // structure per proc data
        std::vector< std::map<index_type, value_type> >
            per_proc_data(_world.size());

        for(size_t i = 0; i < width && _global_col < _ncol;
            _global_col++, i++) {

            // if I don't own anything in this column, continue
            if(_col_itr.colid() + _offset_col != _global_col)
                    continue;

            // first accumulate column values, then distribute
            for(cb_nz_itr_t nz = _data.begnz(_col_itr);
                nz != _data.endnz(_col_itr); nz++) {

                index_type g_cb_col_idx = _col_itr.colid() + _offset_col;
                index_type g_cb_row_idx = nz.rowid()  + _offset_row;
                if(_transp_b) std::swap(g_cb_row_idx, g_cb_col_idx);

                size_t target_proc = A.Owner(g_cb_row_idx, g_cb_col_idx);
                index_type coords = idx(g_cb_row_idx, g_cb_col_idx);
                per_proc_data[target_proc].insert(
                    std::make_pair(coords, nz.value()));
            }

            // this proc can advance to next column
            _col_itr++;
        }

        std::set<index_type> indices;
        _set_view(per_proc_data, indices);
    }

    /**
     *  Extract a Elemental view of CombBLAS ROWS. The distribution of the
     *  Elemental view is determined by the distribution of the input matrix
     *  A.
     *  The row_idxs store the global row indices that contain non-zeros.
     */
    template<typename dist_elem_matrix_t>
    void extract_elemental_row_slab_view(
            const dist_elem_matrix_t &A,
            std::set<index_type> &row_idxs,
            size_t height = 1) {

        _values.clear();

        // structure per proc data
        std::vector< std::map<index_type, value_type> >
            per_proc_data(_world.size());

        // accumulate locale values
        cb_col_itr_t col_itr;
        for(col_itr = _data.begcol(); col_itr != _data.endcol(); col_itr++) {

            // first accumulate column values, then distribute
            size_t r = 0;
            for(; r < height && _row_itr != _data.endnz(col_itr); _row_itr++, r++) {

                index_type g_cb_col_idx = col_itr.colid()  + _offset_col;
                index_type g_cb_row_idx = _row_itr.rowid() + _offset_row;
                if(_transp_b) std::swap(g_cb_row_idx, g_cb_col_idx);

                size_t target_proc = A.Owner(g_cb_row_idx, g_cb_col_idx);
                index_type coords = idx(g_cb_row_idx, g_cb_col_idx);
                per_proc_data[target_proc].insert(
                    std::make_pair(coords, _row_itr.value()));
            }
        }

        std::set<index_type> indices;
        _set_view(per_proc_data, indices);
        typename std::set<index_type>::iterator itr;
        for(itr = indices.begin(); itr != indices.end(); itr++)
            row_idxs.insert(row(*itr));
    }

    /**
     *  Extract a Elemental view of CombBLAS ROWS in sequence. The
     *  distribution of the Elemental view is determined by the distribution
     *  of the input matrix A.
     *  @Note CombBLAS is not well suited for this kind of traversals..
     */
    template<typename dist_elem_matrix_t>
    void extract_elemental_row_slab_view(
            const dist_elem_matrix_t &A,
            size_t height = 1) {

        _values.clear();

        // structure per proc data
        std::vector< std::map<index_type, value_type> >
            per_proc_data(_world.size());

        for(size_t col = 0; col < _row_itrs.size(); col++) {

            cb_nz_itr_t tmp_row_itr = _row_itrs[col];

            for(size_t i = 0; i < height && _global_row + i < _nrow; i++) {

                if(tmp_row_itr.rowid() + _offset_row != _global_row + i)
                    continue;

                index_type g_cb_col_idx = col + _offset_col;
                index_type g_cb_row_idx = tmp_row_itr.rowid() + _offset_row;
                if(_transp_b) std::swap(g_cb_row_idx, g_cb_col_idx);

                size_t target_proc = A.Owner(g_cb_row_idx, g_cb_col_idx);
                index_type coords = idx(g_cb_row_idx, g_cb_col_idx);
                per_proc_data[target_proc].insert(
                    std::make_pair(coords, tmp_row_itr.value()));

                tmp_row_itr++;
                _row_itrs[col]++;
            }
        }

        _global_row += height;

        std::set<index_type> indices;
        _set_view(per_proc_data, indices);
    }

    /**
     *  Extract the same full (redundantly stored) column view of the CombBLAS
     *  matrix on all processors.
     *  @caveat This method uses collective communication.
     */
    void extract_full_slab_view(std::set<index_type> &column_idxs,
                                const size_t slab_size = 1) {

        std::map<index_type, value_type> column_values;
        _values.clear();

        for(size_t i = 0; i < slab_size && _col_itr != _data.endcol();
            _col_itr++, i++) {

            // first accumulate column values, then distribute
            for(cb_nz_itr_t nz = _data.begnz(_col_itr);
                nz != _data.endnz(_col_itr); nz++) {

                index_type g_cb_col_idx = _col_itr.colid() + _offset_col;
                index_type g_cb_row_idx = nz.rowid()  + _offset_row;
                if(_transp_b) std::swap(g_cb_row_idx, g_cb_col_idx);

                index_type coords = idx(g_cb_row_idx, g_cb_col_idx);
                column_values.insert(std::make_pair(coords, nz.value()));
            }
        }

        std::vector< std::map< index_type, value_type> > vector_of_maps;
        boost::mpi::all_gather< std::map<index_type, value_type> > (
            _world, column_values, vector_of_maps);

        // gather vectors in one map
        typename std::map<index_type, value_type>::iterator itr;
        for(size_t i = 0; i < vector_of_maps.size(); ++i) {
            for(itr = vector_of_maps[i].begin();
                itr != vector_of_maps[i].end(); itr++) {

                column_idxs.insert(col(itr->first));

                if(_values.count(itr->first) != 0)
                    _values[itr->first] += itr->second;
                else
                    _values.insert(std::make_pair(itr->first, itr->second));
            }
        }
    }

    void extract_full_slab_view(const size_t width = 1) {

        _values.clear();

        std::map<index_type, value_type> column_values;

        for(size_t i = 0; i < width && _global_col < _ncol;
            _global_col++, i++) {

            // if I don't own anything in this column, continue
            if(_col_itr.colid() + _offset_col != _global_col)
                    continue;

            // first accumulate column values, then distribute
            for(cb_nz_itr_t nz = _data.begnz(_col_itr);
                nz != _data.endnz(_col_itr); nz++) {

                index_type g_cb_col_idx = _col_itr.colid() + _offset_col;
                index_type g_cb_row_idx = nz.rowid()  + _offset_row;
                if(_transp_b) std::swap(g_cb_row_idx, g_cb_col_idx);

                index_type coords = idx(g_cb_row_idx, g_cb_col_idx);
                column_values.insert(std::make_pair(coords, nz.value()));
            }

            // this proc can advance to next column
            _col_itr++;
        }

        std::vector< std::map< index_type, value_type> > vector_of_maps;
        boost::mpi::all_gather< std::map<index_type, value_type> > (
            _world, column_values, vector_of_maps);

        // gather vectors in one map
        typename std::map<index_type, value_type>::iterator itr;
        for(size_t i = 0; i < vector_of_maps.size(); ++i) {
            for(itr = vector_of_maps[i].begin();
                itr != vector_of_maps[i].end(); itr++) {

                if(_values.count(itr->first) != 0)
                    _values[itr->first] += itr->second;
                else
                    _values.insert(std::make_pair(itr->first, itr->second));
            }
        }
    }

    /// get the number of columns
    index_type ncols() { return _ncol; }
    index_type nrows() { return _nrow; }

private:

    /// world communicator of the view
    boost::mpi::communicator _world;

    /// the CombBLAS data
    cb_col_t &_data;

    /// the iterator holding the current position in iterating the CombBLAS
    /// matrix. This is used when generating slabs column wise.
    cb_col_itr_t _col_itr;

    /// the iterator holding the current position in iterating the CombBLAS
    /// matrix. This is used when generating slabs row wise.
    cb_nz_itr_t _row_itr;
    std::vector<cb_nz_itr_t> _row_itrs;

    /// transpose flag
    const bool _transp_b;

    /// this processors row offset in the CombBLAS matrix
    const size_t _offset_row;

    /// this processors column offset in the CombBLAS matrix
    const size_t _offset_col;

    size_t _global_row;
    size_t _global_col;

    /// number of columns
    size_t _ncol;
    /// number of rows
    size_t _nrow;

    /// holds the values of the current slab view
    std::map<index_type, value_type> _values;

    /// convert an 1D index to the global row index
    index_type row(index_type value) { return  value / _ncol; }

    /// convert an 1D index to the global column index
    index_type col(index_type value) { return  value % _ncol; }

    /// convert a (row, col) pair to a 1D index
    index_type idx(index_type row, index_type col) {
        return row * _ncol + col;
    }

    //FIXME: here we use collectives, add existing code for other comm schemes
    /// This methods takes a mapping from indices to processors and
    /// communicates the values to the target processor.
    void _set_view(
            std::vector< std::map<index_type, value_type> > &per_proc_data,
            std::set<index_type> &indices) {

        //// first we need to know how many elements we receive from other
        //// processors:
        //std::vector<size_t> n_values(_world.size());
        //for(size_t idx = 0; idx < _world.size(); ++idx)
            //n_values[idx] = per_proc_data[idx].size();

        //// resulting vector holds expected values: vector_of_sizes[from][to]
        //std::vector< std::vector<size_t> > vector_of_sizes;
        //boost::mpi::all_gather< std::vector<size_t> > (
            //_world, n_values, vector_of_sizes);

        // pre-post receives for all values then
        //FIXME: this is not working because Boost sends serialized data in
        // two steps: first the size and then the data, so waiting for one
        // irecv is not correct.
        //std::vector<boost::mpi::request> reqs;
        //std::vector< std::map<index_type, value_type> > vector_of_maps(_world.size());
        //for(size_t from = 0; from < _world.size(); ++from) {
            //size_t recv_size = vector_of_sizes[from][_world.rank()];
            //std::cout << "from " << from << " receive " << recv_size << std::endl;
            //if(recv_size > 0)
                //reqs.push_back(
                    //_world.irecv(from, 0, &vector_of_maps[from], recv_size));
        //}

        //// then perform all sends
        //for(size_t to = 0; to < _world.size(); ++to) {
            //if(per_proc_data[to].size() > 0)
                //_world.send(to, 0, per_proc_data[to]);
        //}

        //// wait for completion
        //boost::mpi::wait_all(reqs.begin(), reqs.end());

        std::vector<std::map<index_type, value_type> > vector_of_maps;
        for(int i = 0; i < _world.size(); ++i)
            boost::mpi::gather(_world, per_proc_data[i], vector_of_maps, i);

        typename std::map<index_type, value_type>::iterator itr;
        for(size_t i = 0; i < vector_of_maps.size(); ++i) {
            for(itr = vector_of_maps[i].begin();
                itr != vector_of_maps[i].end(); itr++) {

                indices.insert(itr->first);

                if(_values.count(itr->first) != 0)
                    _values[itr->first] += itr->second;
                else
                    _values.insert(std::make_pair(itr->first, itr->second));
            }
        }
    }
};

#endif

} } // namespace skylark::utility

#endif //SKYLARK_COMBBLAS_SLAB_VIEW

max_issues_repo_name: cjiyer/libskylark

max_forks_repo_name: cjiyer/libskylark