streaming/stream_session: Don't stop stream manager

We cannot stop the stream manager because it's accessible via the API
server during shutdown, for example, which can cause a SIGSEGV.

Spotted by ASan.
Message-Id: <1453130811-22540-1-git-send-email-penberg@scylladb.com>

SCYLLA-VERSION-GEN

@@ -1,6 +1,6 @@
 #!/bin/sh
 
-VERSION=666.development
+VERSION=0.16
 
 if test -f version
 then

api/api-doc/messaging_service.json

@@ -253,7 +253,8 @@
                  "STREAM_MUTATION",
                  "STREAM_MUTATION_DONE",
                  "COMPLETE_MESSAGE",
-                 "LAST"
+                 "REPAIR_CHECKSUM_RANGE",
+                 "GET_SCHEMA_VERSION"
                ]
             }
          }

api/messaging_service.cc

@@ -34,7 +34,7 @@ namespace api {
 using shard_info = messaging_service::shard_info;
 using msg_addr = messaging_service::msg_addr;
 
-static const int32_t num_verb = static_cast<int32_t>(messaging_verb::LAST) + 1;
+static const int32_t num_verb = static_cast<int32_t>(messaging_verb::LAST);
 
 std::vector<message_counter> map_to_message_counters(
         const std::unordered_map<gms::inet_address, unsigned long>& map) {
@@ -124,7 +124,7 @@ void set_messaging_service(http_context& ctx, routes& r) {
     });
 
     get_dropped_messages_by_ver.set(r, [](std::unique_ptr<request> req) {
-        shared_ptr<std::vector<uint64_t>> map = make_shared<std::vector<uint64_t>>(num_verb, 0);
+        shared_ptr<std::vector<uint64_t>> map = make_shared<std::vector<uint64_t>>(num_verb);
 
         return net::get_messaging_service().map_reduce([map](const uint64_t* local_map) mutable {
             for (auto i = 0; i < num_verb; i++) {
@@ -137,8 +137,12 @@ void set_messaging_service(http_context& ctx, routes& r) {
             for (auto i : verb_counter::verb_wrapper::all_items()) {
                 verb_counter c;
                 messaging_verb v = i; // for type safety we use messaging_verb values
-                if ((*map)[static_cast<int32_t>(v)] > 0) {
-                    c.count = (*map)[static_cast<int32_t>(v)];
+                auto idx = static_cast<uint32_t>(v);
+                if (idx >= map->size()) {
+                    throw std::runtime_error(sprint("verb index out of bounds: %lu, map size: %lu", idx, map->size()));
+                }
+                if ((*map)[idx] > 0) {
+                    c.count = (*map)[idx];
                     c.verb = i;
                     res.push_back(c);
                 }

cql3/statements/alter_table_statement.cc

@@ -42,6 +42,7 @@
 #include "cql3/statements/alter_table_statement.hh"
 #include "service/migration_manager.hh"
 #include "validation.hh"
+#include "db/config.hh"
 
 namespace cql3 {
 
@@ -77,9 +78,13 @@ void alter_table_statement::validate(distributed<service::storage_proxy>& proxy,
     // validated in announce_migration()
 }
 
+static const sstring ALTER_TABLE_FEATURE = "ALTER TABLE";
+
 future<bool> alter_table_statement::announce_migration(distributed<service::storage_proxy>& proxy, bool is_local_only)
 {
     auto& db = proxy.local().get_db().local();
+    db.get_config().check_experimental(ALTER_TABLE_FEATURE);
+
     auto schema = validation::validate_column_family(db, keyspace(), column_family());
     auto cfm = schema_builder(schema);
 

db/config.cc

@@ -30,6 +30,7 @@
 #include "core/shared_ptr.hh"
 #include "core/fstream.hh"
 #include "core/do_with.hh"
+#include "core/print.hh"
 #include "log.hh"
 #include <boost/any.hpp>
 
@@ -432,3 +433,9 @@ boost::filesystem::path db::config::get_conf_dir() {
 
     return confdir;
 }
+
+void db::config::check_experimental(const sstring& what) const {
+    if (!experimental()) {
+        throw std::runtime_error(sprint("%s is currently disabled. Start Scylla with --experimental=on to enable.", what));
+    }
+}

db/config.hh

@@ -102,6 +102,9 @@ public:
 
     config();
 
+    // Throws exception if experimental feature is disabled.
+    void check_experimental(const sstring& what) const;
+
     boost::program_options::options_description
     get_options_description();
 
@@ -690,7 +693,7 @@ public:
     val(ssl_storage_port, uint32_t, 7001, Used,     \
             "The SSL port for encrypted communication. Unused unless enabled in encryption_options."  \
     )                                                   \
-    val(default_log_level, sstring, "warn", Used, \
+    val(default_log_level, sstring, "info", Used, \
             "Default log level for log messages.  Valid values are trace, debug, info, warn, error.") \
     val(logger_log_level, string_map, /* none */, Used,\
             "map of logger name to log level.  Valid values are trace, debug, info, warn, error.  " \
@@ -717,6 +720,7 @@ public:
     val(ring_delay_ms, uint32_t, 30 * 1000, Used, "Time a node waits to hear from other nodes before joining the ring in milliseconds. Same as -Dcassandra.ring_delay_ms in cassandra.") \
     val(developer_mode, bool, false, Used, "Relax environement checks. Setting to true can reduce performance and reliability significantly.") \
     val(skip_wait_for_gossip_to_settle, int32_t, -1, Used, "An integer to configure the wait for gossip to settle. -1: wait normally, 0: do not wait at all, n: wait for at most n polls. Same as -Dcassandra.skip_wait_for_gossip_to_settle in cassandra.") \
+    val(experimental, bool, false, Used, "Set to true to unlock experimental features.") \
     /* done! */
 
 #define _make_value_member(name, type, deflt, status, desc, ...)    \
@@ -733,5 +737,4 @@ private:
     int _dummy;
 };
 
-
 }

dist/ami/scylla.json

@@ -8,7 +8,7 @@
       "security_group_id": "{{user `security_group_id`}}",
       "region": "{{user `region`}}",
       "associate_public_ip_address": "{{user `associate_public_ip_address`}}",
-      "source_ami": "ami-61bbf104",
+      "source_ami": "ami-8ef1d6e4",
       "user_data_file": "user_data.txt",
       "instance_type": "{{user `instance_type`}}",
       "ssh_username": "centos",

dist/common/scripts/scylla_coredump_setup

@@ -31,8 +31,8 @@ else
 [Coredump]
 Storage=external
 Compress=yes
-ProcessSizeMax=16G
-ExternalSizeMax=16G
+ProcessSizeMax=1024G
+ExternalSizeMax=1024G
 EOS
     if [ $SYMLINK = 1 ]; then
         rm -rf /var/lib/systemd/coredump

dist/common/scripts/scylla_ntp_setup

@@ -32,7 +32,10 @@ else
         sed -e s#fedora.pool.ntp.org#amazon.pool.ntp.org# /etc/ntp.conf > /tmp/ntp.conf
         mv /tmp/ntp.conf /etc/ntp.conf
     fi
-    systemctl enable ntpd.service
+    if [ "`systemctl is-active ntpd`" = "active" ]; then
+        systemctl stop ntpd.service
+    fi
     ntpdate `cat /etc/ntp.conf |grep "^server"|head -n1|awk '{print $2}'`
+    systemctl enable ntpd.service
     systemctl start ntpd.service
 fi

dist/common/scripts/scylla_setup

@@ -39,9 +39,11 @@ SYSCONFIG_SETUP_ARGS="-n $NIC"
 . /etc/os-release
 
 if [ "$ID" != "ubuntu" ]; then
-    setenforce 0
-    sed -e "s/enforcing/disabled/" /etc/sysconfig/selinux > /tmp/selinux
-    mv /tmp/selinux /etc/sysconfig/
+    if [ "`sestatus | awk '{print $3}'`" != "disabled" ]; then
+        setenforce 0
+        sed -e "s/enforcing/disabled/" /etc/sysconfig/selinux > /tmp/selinux
+        mv /tmp/selinux /etc/sysconfig/
+    fi
     if [ $AMI -eq 1 ]; then
         SYSCONFIG_SETUP_ARGS="$SYSCONFIG_SETUP_ARGS -N -a"
         if [ "$LOCAL_PKG" = "" ]; then

dist/common/sudoers.d/scylla

@@ -1 +1 @@
-scylla ALL=(ALL) NOPASSWD: /usr/lib/scylla/scylla_prepare,/usr/lib/scylla/scylla_stop
+scylla ALL=(ALL) NOPASSWD:SETENV: /usr/lib/scylla/scylla_prepare,/usr/lib/scylla/scylla_stop

dist/redhat/systemd/scylla-server.service

@@ -9,9 +9,9 @@ LimitNOFILE=200000
 LimitAS=infinity
 LimitNPROC=8096
 EnvironmentFile=/etc/sysconfig/scylla-server
-ExecStartPre=/usr/bin/sudo /usr/lib/scylla/scylla_prepare
+ExecStartPre=/usr/bin/sudo -E /usr/lib/scylla/scylla_prepare
 ExecStart=/usr/lib/scylla/scylla_run
-ExecStopPost=/usr/bin/sudo /usr/lib/scylla/scylla_stop
+ExecStopPost=/usr/bin/sudo -E /usr/lib/scylla/scylla_stop
 TimeoutStartSec=900
 KillMode=process
 Restart=no

streaming/stream_session.cc

@@ -241,11 +241,12 @@ stream_session::~stream_session() = default;
 
 future<> stream_session::init_streaming_service(distributed<database>& db) {
     _db = &db;
-    engine().at_exit([] {
-        return _handlers.stop().then([]{
-            return get_stream_manager().stop();
-        });
-    });
+    // #293 - do not stop anything
+    // engine().at_exit([] {
+    //     return _handlers.stop().then([]{
+    //         return get_stream_manager().stop();
+    //     });
+    // });
     return get_stream_manager().start().then([] {
         return _handlers.start().then([] {
             return _handlers.invoke_on_all([] (handler& h) {

tests/cql_test_env.cc

@@ -272,6 +272,7 @@ public:
             cfg->commitlog_directory() = _data_dir->path + "/commitlog.dir";
             cfg->num_tokens() = 256;
             cfg->ring_delay_ms() = 500;
+            cfg->experimental() = true;
             boost::filesystem::create_directories((_data_dir->path + "/system").c_str());
             boost::filesystem::create_directories(cfg->commitlog_directory().c_str());
             tst_init_storage_service(*db).get();