Update seastar submodule

* seastar 777ab50...c2489c6 (1):
  > Merge "Fix signal mask corruption" from Tomasz

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "seastar"]
 	path = seastar
-	url = ../seastar
+	url = ../scylla-seastar
 	ignore = dirty
 [submodule "swagger-ui"]
 	path = swagger-ui

SCYLLA-VERSION-GEN

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

dist/docker/redhat/commandlineparser.py

@@ -9,6 +9,7 @@ def parse():
     parser.add_argument('--smp', default=None, help="e.g --smp 2 to use two CPUs")
     parser.add_argument('--memory', default=None, help="e.g. --memory 1G to use 1 GB of RAM")
     parser.add_argument('--overprovisioned', default='0', choices=['0', '1'], help="run in overprovisioned environment")
+    parser.add_argument('--listen-address', default=None, dest='listenAddress')
     parser.add_argument('--broadcast-address', default=None, dest='broadcastAddress')
     parser.add_argument('--broadcast-rpc-address', default=None, dest='broadcastRpcAddress')
     return parser.parse_args()

dist/docker/redhat/scyllasetup.py

@@ -8,6 +8,7 @@ class ScyllaSetup:
         self._developerMode = arguments.developerMode
         self._seeds = arguments.seeds
         self._cpuset = arguments.cpuset
+        self._listenAddress = arguments.listenAddress
         self._broadcastAddress = arguments.broadcastAddress
         self._broadcastRpcAddress = arguments.broadcastRpcAddress
         self._smp = arguments.smp
@@ -31,14 +32,15 @@ class ScyllaSetup:
 
     def scyllaYAML(self):
         configuration = yaml.load(open('/etc/scylla/scylla.yaml'))
-        IP = subprocess.check_output(['hostname', '-i']).decode('ascii').strip()
-        configuration['listen_address'] = IP
-        configuration['rpc_address'] = IP
+        if self._listenAddress is None:
+            self._listenAddress = subprocess.check_output(['hostname', '-i']).decode('ascii').strip()
+        configuration['listen_address'] = self._listenAddress
+        configuration['rpc_address'] = self._listenAddress
         if self._seeds is None:
             if self._broadcastAddress is not None:
                 self._seeds = self._broadcastAddress
             else:
-                self._seeds = IP
+                self._seeds = self._listenAddress
         configuration['seed_provider'] = [
                 {'class_name': 'org.apache.cassandra.locator.SimpleSeedProvider',
                  'parameters': [{'seeds': self._seeds}]}

dist/redhat/scylla.spec.in

@@ -27,7 +27,7 @@ Group:          Applications/Databases
 Summary:        The Scylla database server
 License:        AGPLv3
 URL:            http://www.scylladb.com/
-BuildRequires:  libaio-devel libstdc++-devel cryptopp-devel hwloc-devel numactl-devel libpciaccess-devel libxml2-devel zlib-devel thrift-devel yaml-cpp-devel lz4-devel snappy-devel jsoncpp-devel systemd-devel xz-devel openssl-devel libcap-devel libselinux-devel libgcrypt-devel libgpg-error-devel elfutils-devel krb5-devel libcom_err-devel libattr-devel pcre-devel elfutils-libelf-devel bzip2-devel keyutils-libs-devel xfsprogs-devel make gnutls-devel systemd-devel lksctp-tools-devel protobuf-devel protobuf-compiler
+BuildRequires:  libaio-devel libstdc++-devel cryptopp-devel hwloc-devel numactl-devel libpciaccess-devel libxml2-devel zlib-devel thrift-devel yaml-cpp-devel lz4-devel snappy-devel jsoncpp-devel systemd-devel xz-devel openssl-devel libcap-devel libselinux-devel libgcrypt-devel libgpg-error-devel elfutils-devel krb5-devel libcom_err-devel libattr-devel pcre-devel elfutils-libelf-devel bzip2-devel keyutils-libs-devel xfsprogs-devel make gnutls-devel systemd-devel lksctp-tools-devel protobuf-devel protobuf-compiler libunwind-devel
 %{?fedora:BuildRequires: boost-devel ninja-build ragel antlr3-tool antlr3-C++-devel python3 gcc-c++ libasan libubsan python3-pyparsing dnf-yum}
 %{?rhel:BuildRequires: scylla-libstdc++-static scylla-boost-devel scylla-ninja-build scylla-ragel scylla-antlr3-tool scylla-antlr3-C++-devel python34 scylla-gcc-c++ >= 5.1.1, python34-pyparsing}
 Requires:       scylla-conf systemd-libs hwloc collectd PyYAML python-urwid pciutils pyparsing python-requests curl bc util-linux

dist/ubuntu/build_deb.sh

@@ -78,13 +78,13 @@ cp dist/ubuntu/scylla-server.install.in debian/scylla-server.install
 if [ "$RELEASE" = "14.04" ]; then
     sed -i -e "s/@@DH_INSTALLINIT@@/--upstart-only/g" debian/rules
     sed -i -e "s/@@COMPILER@@/g++-5/g" debian/rules
-    sed -i -e "s/@@BUILD_DEPENDS@@/g++-5/g" debian/control
+    sed -i -e "s/@@BUILD_DEPENDS@@/g++-5, libunwind8-dev/g" debian/control
     sed -i -e "s#@@INSTALL@@#dist/ubuntu/sudoers.d/scylla etc/sudoers.d#g" debian/scylla-server.install
     sed -i -e "s#@@HKDOTTIMER@@##g" debian/scylla-server.install
 else
     sed -i -e "s/@@DH_INSTALLINIT@@//g" debian/rules
     sed -i -e "s/@@COMPILER@@/g++/g" debian/rules
-    sed -i -e "s/@@BUILD_DEPENDS@@/libsystemd-dev, g++/g" debian/control
+    sed -i -e "s/@@BUILD_DEPENDS@@/libsystemd-dev, g++, libunwind-dev/g" debian/control
     sed -i -e "s#@@INSTALL@@##g" debian/scylla-server.install
     sed -i -e "s#@@HKDOTTIMER@@#dist/common/systemd/scylla-housekeeping.timer /lib/systemd/system#g" debian/scylla-server.install
 fi

docs/docker-hub.md

@@ -97,6 +97,16 @@ For example, to configure Scylla to run with two seed nodes `192.168.0.100` and
 $ docker run --name some-scylla -d scylladb/scylla --seeds 192.168.0.100,192.168.0.200
 ```
 
+### `--listen-address ADDR`
+
+The `--listen-address` command line option configures the IP address the Scylla instance listens for client connections.
+
+For example, to configure Scylla to use listen address `10.0.0.5`:
+
+```console
+$ docker run --name some-scylla -d scylladb/scylla --listen-address 10.0.0.5
+```
+
 ### `--broadcast-address ADDR`
 
 The `--broadcast-address` command line option configures the IP address the Scylla instance tells other Scylla nodes in the cluster to connect to.

service/pager/query_pagers.cc

@@ -124,6 +124,43 @@ private:
                 logger.trace("Result ranges {}", ranges);
             };
 
+            // Because of #1446 we don't have a comparator to use with
+            // range<clustering_key_prefix> which would produce correct results.
+            // This means we cannot reuse the same logic for dealing with
+            // partition and clustering keys.
+            auto modify_ck_ranges = [reversed] (const schema& s, auto& ranges, auto& lo) {
+                typedef typename std::remove_reference_t<decltype(ranges)>::value_type range_type;
+                typedef typename range_type::bound bound_type;
+
+                auto cmp = [reversed, bv_cmp = bound_view::compare(s)] (const auto& a, const auto& b) {
+                    return reversed ? bv_cmp(b, a) : bv_cmp(a, b);
+                };
+                auto start_bound = [reversed] (const auto& range) -> const bound_view& {
+                    return reversed ? range.second : range.first;
+                };
+                auto end_bound = [reversed] (const auto& range) -> const bound_view& {
+                    return reversed ? range.first : range.second;
+                };
+                clustering_key_prefix::equality eq(s);
+
+                auto it = ranges.begin();
+                while (it != ranges.end()) {
+                    auto range = bound_view::from_range(*it);
+                    if (cmp(end_bound(range), lo) || eq(end_bound(range).prefix, lo)) {
+                        logger.trace("Remove ck range {}", *it);
+                        it = ranges.erase(it);
+                        continue;
+                    } else if (cmp(start_bound(range), lo)) {
+                        assert(cmp(lo, end_bound(range)));
+                        auto r = reversed ? range_type(it->start(), bound_type { lo, false })
+                                          : range_type(bound_type { lo, false }, it->end());
+                        logger.trace("Modify ck range {} -> {}", *it, r);
+                        *it = std::move(r);
+                    }
+                    ++it;
+                }
+            };
+
             // last ck can be empty depending on whether we
             // deserialized state or not. This case means "last page ended on
             // something-not-bound-by-clustering" (i.e. a static row, alone)
@@ -136,15 +173,7 @@ private:
             if (has_ck) {
                 query::clustering_row_ranges row_ranges = _cmd->slice.default_row_ranges();
                 clustering_key_prefix ckp = clustering_key_prefix::from_exploded(*_schema, _last_ckey->explode(*_schema));
-                clustering_key_prefix::less_compare cmp_rt(*_schema);
-                modify_ranges(row_ranges, ckp, false, [&cmp_rt](auto& c1, auto c2) {
-                    if (cmp_rt(c1, c2)) {
-                        return -1;
-                    } else if (cmp_rt(c2, c1)) {
-                        return 1;
-                    }
-                    return 0;
-                });
+                modify_ck_ranges(*_schema, row_ranges, ckp);
 
                 _cmd->slice.set_range(*_schema, *_last_pkey, row_ranges);
             }

service/storage_proxy.cc

@@ -3452,8 +3452,8 @@ storage_proxy::query_mutations_locally(schema_ptr s, lw_shared_ptr<query::read_c
             });
         });
     } else {
-        return _db.map_reduce(mutation_result_merger{cmd, s}, [cmd, &pr, gs = global_schema_ptr(s), trace_state = std::move(trace_state)] (database& db) {
-            return db.query_mutations(gs, *cmd, pr, trace_state).then([] (reconcilable_result&& result) {
+        return _db.map_reduce(mutation_result_merger{ cmd, s }, [cmd, &pr, gs=global_schema_ptr(s), gt = tracing::global_trace_state_ptr(std::move(trace_state))] (database& db) {
+            return db.query_mutations(gs, *cmd, pr, gt).then([] (reconcilable_result&& result) {
                 return make_foreign(make_lw_shared(std::move(result)));
             });
         }).then([] (reconcilable_result&& result) {

tracing/trace_state.hh

@@ -600,4 +600,62 @@ inline void stop_foreground(const trace_state_ptr& state) {
         state->stop_foreground_and_write();
     }
 }
+
+// global_trace_state_ptr is a helper class that may be used for creating spans
+// of an existing tracing session on other shards. When a tracing span on a
+// different shard is needed global_trace_state_ptr would create a secondary
+// tracing session on that shard similarly to what we do when we create tracing
+// spans on remote Nodes.
+//
+// The usage is straight forward:
+// 1. Create a global_trace_state_ptr from the existing trace_state_ptr object.
+// 2. Pass it to the execution unit that (possibly) runs on a different shard
+//    and pass the global_trace_state_ptr object instead of a trace_state_ptr
+//    object.
+class global_trace_state_ptr {
+    unsigned _cpu_of_origin;
+    trace_state_ptr _ptr;
+public:
+    // Note: the trace_state_ptr must come from the current shard
+    global_trace_state_ptr(trace_state_ptr t)
+            : _cpu_of_origin(engine().cpu_id())
+            , _ptr(std::move(t))
+    { }
+
+    // May be invoked across shards.
+    global_trace_state_ptr(const global_trace_state_ptr& other)
+            : global_trace_state_ptr(other.get())
+    { }
+
+    // May be invoked across shards.
+    global_trace_state_ptr(global_trace_state_ptr&& other)
+            : global_trace_state_ptr(other.get())
+    { }
+
+    global_trace_state_ptr& operator=(const global_trace_state_ptr&) = delete;
+
+    // May be invoked across shards.
+    trace_state_ptr get() const {
+        // optimize the "tracing not enabled" case
+        if (!_ptr) {
+            return nullptr;
+        }
+
+        if (_cpu_of_origin != engine().cpu_id()) {
+            auto opt_trace_info = make_trace_info(_ptr);
+            if (opt_trace_info) {
+                trace_state_ptr new_trace_state = tracing::get_local_tracing_instance().create_session(*opt_trace_info);
+                begin(new_trace_state);
+                return new_trace_state;
+            } else {
+                return nullptr;
+            }
+        }
+
+        return _ptr;
+    }
+
+    // May be invoked across shards.
+    operator trace_state_ptr() const { return get(); }
+};
 }

transport/server.cc

@@ -652,13 +652,13 @@ future<> cql_server::connection::process_request() {
                     f.length, mem_estimate, _server._max_request_size));
         }
 
-        return with_semaphore(_server._memory_available, mem_estimate, [this, length = f.length, flags = f.flags, op, stream, tracing_requested] {
-          return read_and_decompress_frame(length, flags).then([this, flags, op, stream, tracing_requested] (temporary_buffer<char> buf) {
+        return get_units(_server._memory_available, mem_estimate).then([this, length = f.length, flags = f.flags, op, stream, tracing_requested] (semaphore_units<> mem_permit) {
+          return this->read_and_decompress_frame(length, flags).then([this, flags, op, stream, tracing_requested, mem_permit = std::move(mem_permit)] (temporary_buffer<char> buf) mutable {
 
             ++_server._requests_served;
             ++_server._requests_serving;
 
-            with_gate(_pending_requests_gate, [this, flags, op, stream, buf = std::move(buf), tracing_requested] () mutable {
+            with_gate(_pending_requests_gate, [this, flags, op, stream, buf = std::move(buf), tracing_requested, mem_permit = std::move(mem_permit)] () mutable {
                 auto bv = bytes_view{reinterpret_cast<const int8_t*>(buf.begin()), buf.size()};
                 auto cpu = pick_request_cpu();
                 return smp::submit_to(cpu, [this, bv = std::move(bv), op, stream, client_state = _client_state, tracing_requested] () mutable {
@@ -672,7 +672,7 @@ future<> cql_server::connection::process_request() {
                 }).then([this, flags] (auto&& response) {
                     _client_state.merge(response.second);
                     return this->write_response(std::move(response.first), _compression);
-                }).then([buf = std::move(buf)] {
+                }).then([buf = std::move(buf), mem_permit = std::move(mem_permit)] {
                     // Keep buf alive.
                 });
             }).handle_exception([] (std::exception_ptr ex) {