diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml
index e5682777..928fbbc0 100644
--- a/.rubocop_todo.yml
+++ b/.rubocop_todo.yml
@@ -11,12 +11,12 @@
 Metrics/AbcSize:
   Max: 19
 
-# Offense count: 4
+# Offense count: 5
 # Configuration parameters: CountComments, CountAsOne.
 Metrics/ClassLength:
-  Max: 387
+  Max: 237
 
-# Offense count: 6
+# Offense count: 5
 # Configuration parameters: CountComments, CountAsOne, ExcludedMethods, IgnoredMethods.
 Metrics/MethodLength:
   Exclude:
diff --git a/History.md b/History.md
index 6c373b9c..1e1382d9 100644
--- a/History.md
+++ b/History.md
@@ -4,6 +4,8 @@ Dalli Changelog
 Unreleased
 ==========
 
+- Add quiet support for incr, decr, append, depend, and flush (petergoldstein)
+- Additional refactoring to allow reuse of connection behavior (petergoldstein)
 - Fix issue in flush such that it wasn't passing the delay argument to memcached (petergoldstein)
 
 3.1.0
diff --git a/lib/dalli.rb b/lib/dalli.rb
index a2fc672f..6055e662 100644
--- a/lib/dalli.rb
+++ b/lib/dalli.rb
@@ -31,7 +31,7 @@ class NotPermittedMultiOpError < DalliError; end
   class NilObject; end # rubocop:disable Lint/EmptyClass
   NOT_FOUND = NilObject.new
 
-  MULTI_KEY = :dalli_multi
+  QUIET = :dalli_multi
 
   def self.logger
     @logger ||= (rails_logger || default_logger)
@@ -63,6 +63,7 @@ def self.logger=(logger)
 require_relative 'dalli/ring'
 require_relative 'dalli/protocol'
 require_relative 'dalli/protocol/binary'
+require_relative 'dalli/protocol/connection_manager'
 require_relative 'dalli/protocol/response_buffer'
 require_relative 'dalli/protocol/server_config_parser'
 require_relative 'dalli/protocol/ttl_sanitizer'
diff --git a/lib/dalli/client.rb b/lib/dalli/client.rb
index d0cebfce..13f41b19 100644
--- a/lib/dalli/client.rb
+++ b/lib/dalli/client.rb
@@ -49,7 +49,7 @@ class Client
     def initialize(servers = nil, options = {})
       @servers = ::Dalli::ServersArgNormalizer.normalize_servers(servers)
       @options = normalize_options(options)
-      @key_manager = ::Dalli::KeyManager.new(options)
+      @key_manager = ::Dalli::KeyManager.new(@options)
       @ring = nil
     end
 
@@ -58,24 +58,39 @@ def initialize(servers = nil, options = {})
     #
 
     ##
-    # Turn on quiet aka noreply support.
-    # All relevant operations within this block will be effectively
-    # pipelined as Dalli will use 'quiet' operations where possible.
-    # Currently supports the set, add, replace and delete operations.
-    def multi
-      old = Thread.current[::Dalli::MULTI_KEY]
-      Thread.current[::Dalli::MULTI_KEY] = true
-      yield
-    ensure
-      @ring&.pipeline_consume_and_ignore_responses
-      Thread.current[::Dalli::MULTI_KEY] = old
+    # Get the value associated with the key.
+    # If a value is not found, then +nil+ is returned.
+    def get(key, req_options = nil)
+      perform(:get, key, req_options)
     end
 
     ##
-    # Get the value associated with the key.
+    # Gat (get and touch) fetch an item and simultaneously update its expiration time.
+    #
     # If a value is not found, then +nil+ is returned.
-    def get(key, options = nil)
-      perform(:get, key, options)
+    def gat(key, ttl = nil)
+      perform(:gat, key, ttl_or_default(ttl))
+    end
+
+    ##
+    # Touch updates expiration time for a given key.
+    #
+    # Returns true if key exists, otherwise nil.
+    def touch(key, ttl = nil)
+      resp = perform(:touch, key, ttl_or_default(ttl))
+      resp.nil? ? nil : true
+    end
+
+    ##
+    # Get the value and CAS ID associated with the key.  If a block is provided,
+    # value and CAS will be passed to the block.
+    def get_cas(key)
+      (value, cas) = perform(:cas, key)
+      #  TODO: This is odd.  Confirm this is working as expected.
+      value = nil if !value || value == 'Not found'
+      return [value, cas] unless block_given?
+
+      yield value, cas
     end
 
     ##
@@ -97,7 +112,21 @@ def get_multi(*keys)
       end
     end
 
-    CACHE_NILS = { cache_nils: true }.freeze
+    ##
+    # Fetch multiple keys efficiently, including available metadata such as CAS.
+    # If a block is given, yields key/data pairs one a time.  Data is an array:
+    # [value, cas_id]
+    # If no block is given, returns a hash of
+    #   { 'key' => [value, cas_id] }
+    def get_multi_cas(*keys)
+      if block_given?
+        pipelined_getter.process(keys) { |*args| yield(*args) }
+      else
+        {}.tap do |hash|
+          pipelined_getter.process(keys) { |k, data| hash[k] = data }
+        end
+      end
+    end
 
     # Fetch the value associated with the key.
     # If a value is found, then it is returned.
@@ -110,19 +139,11 @@ def get_multi(*keys)
     def fetch(key, ttl = nil, req_options = nil)
       req_options = req_options.nil? ? CACHE_NILS : req_options.merge(CACHE_NILS) if cache_nils
       val = get(key, req_options)
-      if not_found?(val) && block_given?
-        val = yield
-        add(key, val, ttl_or_default(ttl), req_options)
-      end
-      val
-    end
-
-    def not_found?(val)
-      cache_nils ? val == ::Dalli::NOT_FOUND : val.nil?
-    end
+      return val unless block_given? && not_found?(val)
 
-    def cache_nils
-      @options[:cache_nils]
+      new_val = yield
+      add(key, new_val, ttl_or_default(ttl), req_options)
+      new_val
     end
 
     ##
@@ -136,8 +157,8 @@ def cache_nils
     # - nil if the key did not exist.
     # - false if the value was changed by someone else.
     # - true if the value was successfully updated.
-    def cas(key, ttl = nil, options = nil, &block)
-      cas_core(key, false, ttl, options, &block)
+    def cas(key, ttl = nil, req_options = nil, &block)
+      cas_core(key, false, ttl, req_options, &block)
     end
 
     ##
@@ -147,30 +168,78 @@ def cas(key, ttl = nil, options = nil, &block)
     # Returns:
     # - false if the value was changed by someone else.
     # - true if the value was successfully updated.
-    def cas!(key, ttl = nil, options = nil, &block)
-      cas_core(key, true, ttl, options, &block)
+    def cas!(key, ttl = nil, req_options = nil, &block)
+      cas_core(key, true, ttl, req_options, &block)
     end
 
-    def set(key, value, ttl = nil, options = nil)
-      perform(:set, key, value, ttl_or_default(ttl), 0, options)
+    ##
+    # Turn on quiet aka noreply support for a number of
+    # memcached operations.
+    #
+    # All relevant operations within this block will be effectively
+    # pipelined as Dalli will use 'quiet' versions.  The invoked methods
+    # will all return nil, rather than their usual response.  Method
+    # latency will be substantially lower, as the caller will not be
+    # blocking on responses.
+    #
+    # Currently supports storage (set, add, replace, append, prepend),
+    # arithmetic (incr, decr), flush and delete operations.  Use of
+    # unsupported operations inside a block will raise an error.
+    #
+    # Any error replies will be discarded at the end of the block, and
+    # Dalli client methods invoked inside the block will not
+    # have return values
+    def quiet
+      old = Thread.current[::Dalli::QUIET]
+      Thread.current[::Dalli::QUIET] = true
+      yield
+    ensure
+      @ring&.pipeline_consume_and_ignore_responses
+      Thread.current[::Dalli::QUIET] = old
+    end
+    alias multi quiet
+
+    def set(key, value, ttl = nil, req_options = nil)
+      set_cas(key, value, 0, ttl, req_options)
+    end
+
+    ##
+    # Set the key-value pair, verifying existing CAS.
+    # Returns the resulting CAS value if succeeded, and falsy otherwise.
+    def set_cas(key, value, cas, ttl = nil, req_options = nil)
+      perform(:set, key, value, ttl_or_default(ttl), cas, req_options)
     end
 
     ##
     # Conditionally add a key/value pair, if the key does not already exist
     # on the server.  Returns truthy if the operation succeeded.
-    def add(key, value, ttl = nil, options = nil)
-      perform(:add, key, value, ttl_or_default(ttl), options)
+    def add(key, value, ttl = nil, req_options = nil)
+      perform(:add, key, value, ttl_or_default(ttl), req_options)
     end
 
     ##
     # Conditionally add a key/value pair, only if the key already exists
     # on the server.  Returns truthy if the operation succeeded.
-    def replace(key, value, ttl = nil, options = nil)
-      perform(:replace, key, value, ttl_or_default(ttl), 0, options)
+    def replace(key, value, ttl = nil, req_options = nil)
+      replace_cas(key, value, 0, ttl, req_options)
+    end
+
+    ##
+    # Conditionally add a key/value pair, verifying existing CAS, only if the
+    # key already exists on the server.  Returns the new CAS value if the
+    # operation succeeded, or falsy otherwise.
+    def replace_cas(key, value, cas, ttl = nil, req_options = nil)
+      perform(:replace, key, value, ttl_or_default(ttl), cas, req_options)
+    end
+
+    # Delete a key/value pair, verifying existing CAS.
+    # Returns true if succeeded, and falsy otherwise.
+    def delete_cas(key, cas = 0)
+      perform(:delete, key, cas)
     end
 
     def delete(key)
-      perform(:delete, key, 0)
+      delete_cas(key, 0)
     end
 
     ##
@@ -187,16 +256,6 @@ def prepend(key, value)
       perform(:prepend, key, value.to_s)
     end
 
-    ##
-    # Flush the memcached server, at 'delay' seconds in the future.
-    # Delay defaults to zero seconds, which means an immediate flush.
-    ##
-    def flush(delay = 0)
-      ring.servers.map { |s| s.request(:flush, delay) }
-    end
-
-    alias flush_all flush
-
     ##
     # Incr adds the given amount to the counter on the memcached server.
     # Amt must be a positive integer value.
@@ -208,8 +267,10 @@ def flush(delay = 0)
     # Note that the ttl will only apply if the counter does not already
     # exist.  To increase an existing counter and update its TTL, use
     # #cas.
+    #
+    # If the value already exists, it must have been set with raw: true
     def incr(key, amt = 1, ttl = nil, default = nil)
-      raise ArgumentError, "Positive values only: #{amt}" if amt.negative?
+      check_positive!(amt)
 
       perform(:incr, key, amt.to_i, ttl_or_default(ttl), default)
     end
@@ -228,35 +289,31 @@ def incr(key, amt = 1, ttl = nil, default = nil)
     # Note that the ttl will only apply if the counter does not already
     # exist.  To decrease an existing counter and update its TTL, use
     # #cas.
+    #
+    # If the value already exists, it must have been set with raw: true
     def decr(key, amt = 1, ttl = nil, default = nil)
-      raise ArgumentError, "Positive values only: #{amt}" if amt.negative?
+      check_positive!(amt)
 
       perform(:decr, key, amt.to_i, ttl_or_default(ttl), default)
     end
 
     ##
-    # Touch updates expiration time for a given key.
-    #
-    # Returns true if key exists, otherwise nil.
-    def touch(key, ttl = nil)
-      resp = perform(:touch, key, ttl_or_default(ttl))
-      resp.nil? ? nil : true
-    end
-
+    # Flush the memcached server, at 'delay' seconds in the future.
+    # Delay defaults to zero seconds, which means an immediate flush.
     ##
-    # Gat (get and touch) fetch an item and simultaneously update its expiration time.
-    #
-    # If a value is not found, then +nil+ is returned.
-    def gat(key, ttl = nil)
-      perform(:gat, key, ttl_or_default(ttl))
+    def flush(delay = 0)
+      ring.servers.map { |s| s.request(:flush, delay) }
     end
+    alias flush_all flush
+
+    ALLOWED_STAT_KEYS = %i[items slabs settings].freeze
 
     ##
     # Collect the stats for each server.
     # You can optionally pass a type including :items, :slabs or :settings to get specific stats
     # Returns a hash like { 'hostname:port' => { 'stat1' => 'value1', ... }, 'hostname2:port' => { ... } }
     def stats(type = nil)
-      type = nil unless [nil, :items, :slabs, :settings].include? type
+      type = nil unless ALLOWED_STAT_KEYS.include? type
       values = {}
       ring.servers.each do |server|
         values[server.name.to_s] = server.alive? ? server.request(:stats, type.to_s) : nil
@@ -272,12 +329,6 @@ def reset_stats
       end
     end
 
-    ##
-    ## Make sure memcache servers are alive, or raise an Dalli::RingError
-    def alive!
-      ring.server_for_key('')
-    end
-
     ##
     ## Version of the memcache servers.
     def version
@@ -289,55 +340,9 @@ def version
     end
 
     ##
-    # Get the value and CAS ID associated with the key.  If a block is provided,
-    # value and CAS will be passed to the block.
-    def get_cas(key)
-      (value, cas) = perform(:cas, key)
-      value = nil if !value || value == 'Not found'
-      if block_given?
-        yield value, cas
-      else
-        [value, cas]
-      end
-    end
-
-    ##
-    # Fetch multiple keys efficiently, including available metadata such as CAS.
-    # If a block is given, yields key/data pairs one a time.  Data is an array:
-    # [value, cas_id]
-    # If no block is given, returns a hash of
-    #   { 'key' => [value, cas_id] }
-    def get_multi_cas(*keys)
-      if block_given?
-        pipelined_getter.process(keys) { |*args| yield(*args) }
-      else
-        {}.tap do |hash|
-          pipelined_getter.process(keys) { |k, data| hash[k] = data }
-        end
-      end
-    end
-
-    ##
-    # Set the key-value pair, verifying existing CAS.
-    # Returns the resulting CAS value if succeeded, and falsy otherwise.
-    def set_cas(key, value, cas, ttl = nil, options = nil)
-      ttl ||= @options[:expires_in].to_i
-      perform(:set, key, value, ttl, cas, options)
-    end
-
-    ##
-    # Conditionally add a key/value pair, verifying existing CAS, only if the
-    # key already exists on the server.  Returns the new CAS value if the
-    # operation succeeded, or falsy otherwise.
-    def replace_cas(key, value, cas, ttl = nil, options = nil)
-      ttl ||= @options[:expires_in].to_i
-      perform(:replace, key, value, ttl, cas, options)
-    end
-
-    # Delete a key/value pair, verifying existing CAS.
-    # Returns true if succeeded, and falsy otherwise.
-    def delete_cas(key, cas = 0)
-      perform(:delete, key, cas)
+    ## Make sure memcache servers are alive, or raise an Dalli::RingError
+    def alive!
+      ring.server_for_key('')
     end
 
     ##
@@ -349,6 +354,16 @@ def close
     end
     alias reset close
 
+    CACHE_NILS = { cache_nils: true }.freeze
+
+    def not_found?(val)
+      cache_nils ? val == ::Dalli::NOT_FOUND : val.nil?
+    end
+
+    def cache_nils
+      @options[:cache_nils]
+    end
+
     # Stub method so a bare Dalli client can pretend to be a connection pool.
     def with
       yield self
@@ -356,15 +371,23 @@ def with
 
     private
 
-    def cas_core(key, always_set, ttl = nil, options = nil)
+    def check_positive!(amt)
+      raise ArgumentError, "Positive values only: #{amt}" if amt.negative?
+    end
+
+    def cas_core(key, always_set, ttl = nil, req_options = nil)
       (value, cas) = perform(:cas, key)
       value = nil if !value || value == 'Not found'
       return if value.nil? && !always_set
 
       newvalue = yield(value)
-      perform(:set, key, newvalue, ttl_or_default(ttl), cas, options)
+      perform(:set, key, newvalue, ttl_or_default(ttl), cas, req_options)
     end
 
+    ##
+    # Uses the argument TTL or the client-wide default.  Ensures
+    # that the value is an integer
+    ##
     def ttl_or_default(ttl)
       (ttl || @options[:expires_in]).to_i
     rescue NoMethodError
@@ -385,7 +408,16 @@ def protocol_implementation
       @protocol_implementation ||= @options.fetch(:protocol_implementation, Dalli::Protocol::Binary)
     end
 
-    # Chokepoint method for instrumentation
+    ##
+    # Chokepoint method for memcached methods with a key argument.
+    # Validates the key, resolves the key to the appropriate server
+    # instance, and invokes the memcached method on the appropriate
+    # server.
+    #
+    # This method also forces retries on network errors - when
+    # a particular memcached instance becomes unreachable, or the
+    # operational times out.
+    ##
     def perform(*all_args)
       return yield if block_given?
 
diff --git a/lib/dalli/options.rb b/lib/dalli/options.rb
index 1e33b6a6..a48cd4ba 100644
--- a/lib/dalli/options.rb
+++ b/lib/dalli/options.rb
@@ -31,19 +31,19 @@ def close
       end
     end
 
-    def pipeline_response_start
+    def pipeline_response_setup
       @lock.synchronize do
         super
       end
     end
 
-    def process_outstanding_pipeline_requests
+    def pipeline_next_responses
       @lock.synchronize do
         super
       end
     end
 
-    def pipeline_response_abort
+    def pipeline_abort
       @lock.synchronize do
         super
       end
diff --git a/lib/dalli/pipelined_getter.rb b/lib/dalli/pipelined_getter.rb
index 31d870ac..801885cb 100644
--- a/lib/dalli/pipelined_getter.rb
+++ b/lib/dalli/pipelined_getter.rb
@@ -19,13 +19,7 @@ def process(keys, &block)
       @ring.lock do
         servers = setup_requests(keys)
         start_time = Time.now
-        loop do
-          # Remove any servers which are not connected
-          servers.delete_if { |s| !s.connected? }
-          break if servers.empty?
-
-          servers = fetch_responses(servers, start_time, @ring.socket_timeout, &block)
-        end
+        servers = fetch_responses(servers, start_time, @ring.socket_timeout, &block) until servers.empty?
       end
     rescue NetworkError => e
       Dalli.logger.debug { e.inspect }
@@ -44,6 +38,10 @@ def setup_requests(keys)
     ##
     # Loop through the server-grouped sets of keys, writing
     # the corresponding getkq requests to the appropriate servers
+    #
+    # It's worth noting that we could potentially reduce bytes
+    # on the wire by switching from getkq to getq, and using
+    # the opaque value to match requests to responses.
     ##
     def make_getkq_requests(groups)
       groups.each do |server, keys_for_server|
@@ -80,7 +78,7 @@ def finish_queries(servers)
     end
 
     def finish_query_for_server(server)
-      server.pipeline_response_start
+      server.pipeline_response_setup
     rescue Dalli::NetworkError
       raise
     rescue Dalli::DalliError => e
@@ -91,10 +89,14 @@ def finish_query_for_server(server)
 
     # Swallows Dalli::NetworkError
     def abort_without_timeout(servers)
-      servers.each(&:pipeline_response_abort)
+      servers.each(&:pipeline_abort)
     end
 
     def fetch_responses(servers, start_time, timeout, &block)
+      # Remove any servers which are not connected
+      servers.delete_if { |s| !s.connected? }
+      return [] if servers.empty?
+
       time_left = remaining_time(start_time, timeout)
       readable_servers = servers_with_response(servers, time_left)
       if readable_servers.empty?
@@ -135,11 +137,11 @@ def abort_with_timeout(servers)
     # Processes responses from a server.  Returns true if there are no
     # additional responses from this server.
     def process_server(server)
-      server.process_outstanding_pipeline_requests.each_pair do |key, value_list|
+      server.pipeline_next_responses.each_pair do |key, value_list|
         yield @key_manager.key_without_namespace(key), value_list
       end
 
-      server.pipeline_response_completed?
+      server.pipeline_complete?
     end
 
     def servers_with_response(servers, timeout)
diff --git a/lib/dalli/protocol/binary.rb b/lib/dalli/protocol/binary.rb
index fbc453e6..5e412ec3 100644
--- a/lib/dalli/protocol/binary.rb
+++ b/lib/dalli/protocol/binary.rb
@@ -1,6 +1,5 @@
 # frozen_string_literal: true
 
-require 'English'
 require 'forwardable'
 require 'socket'
 require 'timeout'
@@ -20,62 +19,30 @@ module Protocol
     class Binary
       extend Forwardable
 
-      attr_accessor :hostname, :port, :weight, :options
-      attr_reader :sock, :socket_type
+      attr_accessor :weight, :options
 
       def_delegators :@value_marshaller, :serializer, :compressor, :compression_min_size, :compress_by_default?
+      def_delegators :@connection_manager, :name, :sock, :hostname, :port, :close, :connected?, :socket_timeout,
+                     :socket_type, :up!, :down!, :write, :reconnect_down_server?, :raise_down_error
 
-      DEFAULTS = {
-        # seconds between trying to contact a remote server
-        down_retry_delay: 30,
-        # connect/read/write timeout for socket operations
-        socket_timeout: 1,
-        # times a socket operation may fail before considering the server dead
-        socket_max_failures: 2,
-        # amount of time to sleep between retries when a failure occurs
-        socket_failure_delay: 0.1
-      }.freeze
-
-      def initialize(attribs, options = {})
-        @hostname, @port, @weight, @socket_type, options = ServerConfigParser.parse(attribs, options)
-        @options = DEFAULTS.merge(options)
+      def initialize(attribs, client_options = {})
+        hostname, port, socket_type, @weight, user_creds = ServerConfigParser.parse(attribs)
+        @options = client_options.merge(user_creds)
         @value_marshaller = ValueMarshaller.new(@options)
-        @response_processor = ResponseProcessor.new(self, @value_marshaller)
-        @response_buffer = ResponseBuffer.new(self, @response_processor)
-
-        reset_down_info
-        @sock = nil
-        @pid = nil
-        @request_in_progress = false
-      end
-
-      def response_buffer
-        @response_buffer ||= ResponseBuffer.new(self, @response_processor)
-      end
-
-      def name
-        if socket_type == :unix
-          hostname
-        else
-          "#{hostname}:#{port}"
-        end
+        @connection_manager = ConnectionManager.new(hostname, port, socket_type, @options)
+        @response_processor = ResponseProcessor.new(@connection_manager, @value_marshaller)
       end
 
       # Chokepoint method for error handling and ensuring liveness
       def request(opkey, *args)
         verify_state(opkey)
-        # The alive? call has the side effect of connecting the underlying
-        # socket if it is not connected, or there's been a disconnect
-        # because of timeout or other error.  Method raises an error
-        # if it can't connect
-        raise_memcached_down_err unless alive?
 
         begin
           send(opkey, *args)
         rescue Dalli::MarshalError => e
-          log_marshall_err(args.first, e)
+          log_marshal_err(args.first, e)
           raise
-        rescue Dalli::DalliError, Dalli::NetworkError, Dalli::ValueOverMaxSize, Timeout::Error
+        rescue Dalli::DalliError
           raise
         rescue StandardError => e
           log_unexpected_err(e)
@@ -83,63 +50,17 @@ def request(opkey, *args)
         end
       end
 
-      def raise_memcached_down_err
-        raise Dalli::NetworkError,
-              "#{name} is down: #{@error} #{@msg}. If you are sure it is running, "\
-              "ensure memcached version is > #{::Dalli::MIN_SUPPORTED_MEMCACHED_VERSION}."
-      end
-
-      def log_marshall_err(key, err)
-        Dalli.logger.error "Marshalling error for key '#{key}': #{err.message}"
-        Dalli.logger.error 'You are trying to cache a Ruby object which cannot be serialized to memcached.'
-      end
-
-      def log_unexpected_err(err)
-        Dalli.logger.error "Unexpected exception during Dalli request: #{err.class.name}: #{err.message}"
-        Dalli.logger.error err.backtrace.join("\n\t")
-      end
-
-      # The socket connection to the underlying server is initialized as a side
-      # effect of this call.  In fact, this is the ONLY place where that
-      # socket connection is initialized.
+      ##
+      # Boolean method used by clients of this class to determine if this
+      # particular memcached instance is available for use.
       def alive?
-        return true if @sock
-        return false unless reconnect_down_server?
-
-        connect
-        !!@sock
+        ensure_connected!
       rescue Dalli::NetworkError
+        # ensure_connected! raises a NetworkError if connection fails.  We
+        # want to capture that error and convert it to a boolean value here.
         false
       end
 
-      def reconnect_down_server?
-        return true unless @last_down_at
-
-        time_to_next_reconnect = @last_down_at + options[:down_retry_delay] - Time.now
-        return true unless time_to_next_reconnect.positive?
-
-        Dalli.logger.debug do
-          format('down_retry_delay not reached for %<name>s (%<time>.3f seconds left)', name: name,
-                                                                                        time: time_to_next_reconnect)
-        end
-        false
-      end
-
-      # Closes the underlying socket and cleans up
-      # socket state.
-      def close
-        return unless @sock
-
-        begin
-          @sock.close
-        rescue StandardError
-          nil
-        end
-        @sock = nil
-        @pid = nil
-        abort_request!
-      end
-
       def lock!; end
 
       def unlock!; end
@@ -149,224 +70,125 @@ def unlock!; end
       # flushing responses for kv pairs that were found.
       #
       # Returns nothing.
-      def pipeline_response_start
+      def pipeline_response_setup
         verify_state(:getkq)
         write_noop
         response_buffer.reset
-        start_request!
-      end
-
-      # Did the last call to #pipeline_response_start complete successfully?
-      def pipeline_response_completed?
-        response_buffer.completed?
-      end
-
-      def pipeline_response(bytes_to_advance = 0)
-        response_buffer.process_single_response(bytes_to_advance)
-      end
-
-      def reconnect_on_pipeline_complete!
-        reconnect! 'multi_response has completed' if pipeline_response_completed?
+        @connection_manager.start_request!
       end
 
       # Attempt to receive and parse as many key/value pairs as possible
-      # from this server. After #pipeline_response_start, this should be invoked
+      # from this server. After #pipeline_response_setup, this should be invoked
       # repeatedly whenever this server's socket is readable until
-      # #pipeline_response_completed?.
+      # #pipeline_complete?.
       #
       # Returns a Hash of kv pairs received.
-      def process_outstanding_pipeline_requests
+      def pipeline_next_responses
         reconnect_on_pipeline_complete!
         values = {}
 
         response_buffer.read
 
-        bytes_to_advance, status, key, value, cas = pipeline_response
-        # Loop while we have at least a complete header in the buffer
-        while bytes_to_advance.positive?
-          # If the status and key length are both zero, then this is the response
+        resp_header, key, value = pipeline_response
+        # resp_header is not nil only if we have a full response to parse
+        # in the buffer
+        while resp_header
+          # If the status is ok and key is nil, then this is the response
           # to the noop at the end of the pipeline
-          if status.zero? && key.nil?
-            finish_pipeline
-            break
-          end
+          finish_pipeline && break if resp_header.ok? && key.nil?
 
-          # If the status is zero and the key len is positive, then this is a
+          # If the status is ok and the key is not nil, then this is a
           # getkq response with a value that we want to set in the response hash
-          values[key] = [value, cas] unless key.nil?
+          values[key] = [value, resp_header.cas] unless key.nil?
 
-          # Get the next set of bytes from the buffer
-          bytes_to_advance, status, key, value, cas = pipeline_response(bytes_to_advance)
+          # Get the next response from the buffer
+          resp_header, key, value = pipeline_response
         end
 
         values
       rescue SystemCallError, Timeout::Error, EOFError => e
-        failure!(e)
-      end
-
-      def read_nonblock
-        @sock.read_available
-      end
-
-      # Called after the noop response is received at the end of a set
-      # of pipelined gets
-      def finish_pipeline
-        response_buffer.clear
-        finish_request!
+        @connection_manager.error_on_request!(e)
       end
 
-      # Abort an earlier #pipeline_response_start. Used to signal an external
-      # timeout. The underlying socket is disconnected, and the exception is
-      # swallowed.
+      # Abort current pipelined get. Generally used to signal an external
+      # timeout during pipelined get.  The underlying socket is
+      # disconnected, and the exception is swallowed.
       #
       # Returns nothing.
-      def pipeline_response_abort
+      def pipeline_abort
         response_buffer.clear
-        abort_request!
-        return true unless @sock
+        @connection_manager.abort_request!
+        return true unless connected?
 
-        failure!(RuntimeError.new('External timeout'))
+        # Closes the connection, which ensures that our connection
+        # is in a clean state for future requests
+        @connection_manager.error_on_request!('External timeout')
       rescue NetworkError
         true
       end
 
-      def read(count)
-        start_request!
-        data = @sock.readfull(count)
-        finish_request!
-        data
-      rescue SystemCallError, Timeout::Error, EOFError => e
-        failure!(e)
+      # Did the last call to #pipeline_response_setup complete successfully?
+      def pipeline_complete?
+        !response_buffer.in_progress?
       end
 
-      def write(bytes)
-        start_request!
-        result = @sock.write(bytes)
-        finish_request!
-        result
-      rescue SystemCallError, Timeout::Error => e
-        failure!(e)
+      def username
+        @options[:username] || ENV['MEMCACHE_USERNAME']
       end
 
-      def connected?
-        !@sock.nil?
+      def password
+        @options[:password] || ENV['MEMCACHE_PASSWORD']
       end
 
-      def socket_timeout
-        @socket_timeout ||= @options[:socket_timeout]
+      def require_auth?
+        !username.nil?
       end
 
       # NOTE: Additional public methods should be overridden in Dalli::Threadsafe
 
       private
 
-      def request_in_progress?
-        @request_in_progress
-      end
-
-      def start_request!
-        @request_in_progress = true
-      end
-
-      def finish_request!
-        @request_in_progress = false
-      end
-
-      def abort_request!
-        @request_in_progress = false
-      end
-
+      ##
+      # Checks to see if we can execute the specified operation.  Checks
+      # whether the connection is in use, and whether the command is allowed
+      ##
       def verify_state(opkey)
-        failure!(RuntimeError.new('Already writing to socket')) if request_in_progress?
-        reconnect_on_fork if fork_detected?
-        verify_allowed_multi!(opkey) if multi?
-      end
-
-      def fork_detected?
-        @pid && @pid != Process.pid
-      end
-
-      ALLOWED_MULTI_OPS = %i[add addq delete deleteq replace replaceq set setq noop].freeze
-      def verify_allowed_multi!(opkey)
-        return if ALLOWED_MULTI_OPS.include?(opkey)
-
-        raise Dalli::NotPermittedMultiOpError, "The operation #{opkey} is not allowed in a multi block."
-      end
-
-      def reconnect_on_fork
-        message = 'Fork detected, re-connecting child process...'
-        Dalli.logger.info { message }
-        reconnect! message
-      end
-
-      # Marks the server instance as needing reconnect.  Raises a
-      # Dalli::NetworkError with the specified message.  Calls close
-      # to clean up socket state
-      def reconnect!(message)
-        close
-        sleep(options[:socket_failure_delay]) if options[:socket_failure_delay]
-        raise Dalli::NetworkError, message
-      end
-
-      # Raises Dalli::NetworkError
-      def failure!(exception)
-        message = "#{name} failed (count: #{@fail_count}) #{exception.class}: #{exception.message}"
-        Dalli.logger.warn { message }
-
-        @fail_count += 1
-        if @fail_count >= options[:socket_max_failures]
-          down!
-        else
-          reconnect! 'Socket operation failed, retrying...'
-        end
-      end
-
-      # Marks the server instance as down.  Updates the down_at state
-      # and raises an Dalli::NetworkError that includes the underlying
-      # error in the message.  Calls close to clean up socket state
-      def down!
-        close
-        log_down_detected
+        @connection_manager.confirm_ready!
+        verify_allowed_quiet!(opkey) if quiet?
 
-        @error = $ERROR_INFO&.class&.name
-        @msg ||= $ERROR_INFO&.message
-        raise Dalli::NetworkError, "#{name} is down: #{@error} #{@msg}"
-      end
-
-      def log_down_detected
-        @last_down_at = Time.now
-
-        if @down_at
-          time = Time.now - @down_at
-          Dalli.logger.debug { format('%<name>s is still down (for %<time>.3f seconds now)', name: name, time: time) }
-        else
-          @down_at = @last_down_at
-          Dalli.logger.warn("#{name} is down")
-        end
+        # The ensure_connected call has the side effect of connecting the
+        # underlying socket if it is not connected, or there's been a disconnect
+        # because of timeout or other error.  Method raises an error
+        # if it can't connect
+        raise_down_error unless ensure_connected!
       end
 
-      def log_up_detected
-        return unless @down_at
+      # The socket connection to the underlying server is initialized as a side
+      # effect of this call.  In fact, this is the ONLY place where that
+      # socket connection is initialized.
+      #
+      # Both this method and connect need to be in this class so we can do auth
+      # as required
+      #
+      # Since this is invoked exclusively in verify_state!, we don't need to worry about
+      # thread safety.  Using it elsewhere may require revisiting that assumption.
+      def ensure_connected!
+        return true if connected?
+        return false unless reconnect_down_server?
 
-        time = Time.now - @down_at
-        Dalli.logger.warn { format('%<name>s is back (downtime was %<time>.3f seconds)', name: name, time: time) }
+        connect # This call needs to be in this class so we can do auth
+        connected?
       end
 
-      def up!
-        log_up_detected
-        reset_down_info
-      end
+      ALLOWED_QUIET_OPS = %i[add replace set delete incr decr append prepend flush noop].freeze
+      def verify_allowed_quiet!(opkey)
+        return if ALLOWED_QUIET_OPS.include?(opkey)
 
-      def reset_down_info
-        @fail_count = 0
-        @down_at = nil
-        @last_down_at = nil
-        @msg = nil
-        @error = nil
+        raise Dalli::NotPermittedMultiOpError, "The operation #{opkey} is not allowed in a quiet block."
       end
 
-      def multi?
-        Thread.current[::Dalli::MULTI_KEY]
+      def quiet?
+        Thread.current[::Dalli::QUIET]
       end
 
       def cache_nils?(opts)
@@ -375,39 +197,52 @@ def cache_nils?(opts)
         opts[:cache_nils] ? true : false
       end
 
+      # Retrieval Commands
       def get(key, options = nil)
         req = RequestFormatter.standard_request(opkey: :get, key: key)
         write(req)
         @response_processor.generic_response(unpack: true, cache_nils: cache_nils?(options))
       end
 
-      def pipelined_get(keys)
-        req = +''
-        keys.each do |key|
-          req << RequestFormatter.standard_request(opkey: :getkq, key: key)
-        end
-        # Could send noop here instead of in pipeline_response_start
+      def gat(key, ttl, options = nil)
+        ttl = TtlSanitizer.sanitize(ttl)
+        req = RequestFormatter.standard_request(opkey: :gat, key: key, ttl: ttl)
         write(req)
+        @response_processor.generic_response(unpack: true, cache_nils: cache_nils?(options))
       end
 
+      def touch(key, ttl)
+        ttl = TtlSanitizer.sanitize(ttl)
+        write(RequestFormatter.standard_request(opkey: :touch, key: key, ttl: ttl))
+        @response_processor.generic_response
+      end
+
+      # TODO: This is confusing, as there's a cas command in memcached
+      # and this isn't it.  Maybe rename?  Maybe eliminate?
+      def cas(key)
+        req = RequestFormatter.standard_request(opkey: :get, key: key)
+        write(req)
+        @response_processor.data_cas_response
+      end
+
+      # Storage Commands
       def set(key, value, ttl, cas, options)
-        opkey = multi? ? :setq : :set
-        process_value_req(opkey, key, value, ttl, cas, options)
+        opkey = quiet? ? :setq : :set
+        storage_req(opkey, key, value, ttl, cas, options)
       end
 
       def add(key, value, ttl, options)
-        opkey = multi? ? :addq : :add
-        cas = 0
-        process_value_req(opkey, key, value, ttl, cas, options)
+        opkey = quiet? ? :addq : :add
+        storage_req(opkey, key, value, ttl, 0, options)
       end
 
       def replace(key, value, ttl, cas, options)
-        opkey = multi? ? :replaceq : :replace
-        process_value_req(opkey, key, value, ttl, cas, options)
+        opkey = quiet? ? :replaceq : :replace
+        storage_req(opkey, key, value, ttl, cas, options)
       end
 
       # rubocop:disable Metrics/ParameterLists
-      def process_value_req(opkey, key, value, ttl, cas, options)
+      def storage_req(opkey, key, value, ttl, cas, options)
         (value, bitflags) = @value_marshaller.store(key, value, options)
         ttl = TtlSanitizer.sanitize(ttl)
 
@@ -415,21 +250,42 @@ def process_value_req(opkey, key, value, ttl, cas, options)
                                                 value: value, bitflags: bitflags,
                                                 ttl: ttl, cas: cas)
         write(req)
-        @response_processor.cas_response unless multi?
+        @response_processor.storage_response unless quiet?
       end
       # rubocop:enable Metrics/ParameterLists
 
+      def append(key, value)
+        opkey = quiet? ? :appendq : :append
+        write_append_prepend opkey, key, value
+      end
+
+      def prepend(key, value)
+        opkey = quiet? ? :prependq : :prepend
+        write_append_prepend opkey, key, value
+      end
+
+      def write_append_prepend(opkey, key, value)
+        write(RequestFormatter.standard_request(opkey: opkey, key: key, value: value))
+        @response_processor.no_body_response unless quiet?
+      end
+
+      # Delete Commands
       def delete(key, cas)
-        opkey = multi? ? :deleteq : :delete
+        opkey = quiet? ? :deleteq : :delete
         req = RequestFormatter.standard_request(opkey: opkey, key: key, cas: cas)
         write(req)
-        @response_processor.generic_response unless multi?
+        @response_processor.no_body_response unless quiet?
       end
 
-      def flush(ttl = 0)
-        req = RequestFormatter.standard_request(opkey: :flush, ttl: ttl)
-        write(req)
-        @response_processor.generic_response
+      # Arithmetic Commands
+      def decr(key, count, ttl, initial)
+        opkey = quiet? ? :decrq : :decr
+        decr_incr opkey, key, count, ttl, initial
+      end
+
+      def incr(key, count, ttl, initial)
+        opkey = quiet? ? :incrq : :incr
+        decr_incr opkey, key, count, ttl, initial
       end
 
       # This allows us to special case a nil initial value, and
@@ -444,29 +300,14 @@ def decr_incr(opkey, key, count, ttl, initial)
         initial ||= 0
         write(RequestFormatter.decr_incr_request(opkey: opkey, key: key,
                                                  count: count, initial: initial, expiry: expiry))
-        @response_processor.decr_incr_response
+        @response_processor.decr_incr_response unless quiet?
       end
 
-      def decr(key, count, ttl, initial)
-        decr_incr :decr, key, count, ttl, initial
-      end
-
-      def incr(key, count, ttl, initial)
-        decr_incr :incr, key, count, ttl, initial
-      end
-
-      def write_append_prepend(opkey, key, value)
-        write_generic RequestFormatter.standard_request(opkey: opkey, key: key, value: value)
-      end
-
-      def write_generic(bytes)
-        write(bytes)
-        @response_processor.generic_response
-      end
-
-      def write_noop
-        req = RequestFormatter.standard_request(opkey: :noop)
-        write(req)
+      # Other Commands
+      def flush(ttl = 0)
+        opkey = quiet? ? :flushq : :flush
+        write(RequestFormatter.standard_request(opkey: opkey, ttl: ttl))
+        @response_processor.no_body_response unless quiet?
       end
 
       # Noop is a keepalive operation but also used to demarcate the end of a set of pipelined commands.
@@ -476,14 +317,6 @@ def noop
         @response_processor.multi_with_keys_response
       end
 
-      def append(key, value)
-        write_append_prepend :append, key, value
-      end
-
-      def prepend(key, value)
-        write_append_prepend :prepend, key, value
-      end
-
       def stats(info = '')
         req = RequestFormatter.standard_request(opkey: :stat, key: info)
         write(req)
@@ -491,66 +324,67 @@ def stats(info = '')
       end
 
       def reset_stats
-        write_generic RequestFormatter.standard_request(opkey: :stat, key: 'reset')
+        write(RequestFormatter.standard_request(opkey: :stat, key: 'reset'))
+        @response_processor.generic_response
       end
 
-      def cas(key)
-        req = RequestFormatter.standard_request(opkey: :get, key: key)
-        write(req)
-        @response_processor.data_cas_response
+      def version
+        write(RequestFormatter.standard_request(opkey: :version))
+        @response_processor.generic_response
       end
 
-      def version
-        write_generic RequestFormatter.standard_request(opkey: :version)
+      def write_noop
+        req = RequestFormatter.standard_request(opkey: :noop)
+        write(req)
       end
 
-      def touch(key, ttl)
-        ttl = TtlSanitizer.sanitize(ttl)
-        write_generic RequestFormatter.standard_request(opkey: :touch, key: key, ttl: ttl)
+      def connect
+        @connection_manager.establish_connection
+        authenticate_connection if require_auth?
+        @version = version # Connect socket if not authed
+        up!
+      rescue Dalli::DalliError
+        raise
       end
 
-      def gat(key, ttl, options = nil)
-        ttl = TtlSanitizer.sanitize(ttl)
-        req = RequestFormatter.standard_request(opkey: :gat, key: key, ttl: ttl)
+      def pipelined_get(keys)
+        req = +''
+        keys.each do |key|
+          req << RequestFormatter.standard_request(opkey: :getkq, key: key)
+        end
+        # Could send noop here instead of in pipeline_response_setup
         write(req)
-        @response_processor.generic_response(unpack: true, cache_nils: cache_nils?(options))
       end
 
-      def connect
-        Dalli.logger.debug { "Dalli::Server#connect #{name}" }
+      def response_buffer
+        @response_buffer ||= ResponseBuffer.new(@connection_manager, @response_processor)
+      end
 
-        begin
-          @pid = Process.pid
-          @sock = memcached_socket
-          authenticate_connection if require_auth?
-          @version = version # Connect socket if not authed
-          up!
-        rescue Dalli::DalliError # SASL auth failure
-          raise
-        rescue SystemCallError, Timeout::Error, EOFError, SocketError => e
-          # SocketError = DNS resolution failure
-          failure!(e)
-        end
+      def pipeline_response
+        response_buffer.process_single_getk_response
       end
 
-      def memcached_socket
-        if socket_type == :unix
-          Dalli::Socket::UNIX.open(hostname, options)
-        else
-          Dalli::Socket::TCP.open(hostname, port, options)
-        end
+      # Called after the noop response is received at the end of a set
+      # of pipelined gets
+      def finish_pipeline
+        response_buffer.clear
+        @connection_manager.finish_request!
+
+        true # to simplify response
       end
 
-      def require_auth?
-        !username.nil?
+      def reconnect_on_pipeline_complete!
+        @connection_manager.reconnect! 'pipelined get has completed' if pipeline_complete?
       end
 
-      def username
-        @options[:username] || ENV['MEMCACHE_USERNAME']
+      def log_marshal_err(key, err)
+        Dalli.logger.error "Marshalling error for key '#{key}': #{err.message}"
+        Dalli.logger.error 'You are trying to cache a Ruby object which cannot be serialized to memcached.'
       end
 
-      def password
-        @options[:password] || ENV['MEMCACHE_PASSWORD']
+      def log_unexpected_err(err)
+        Dalli.logger.error "Unexpected exception during Dalli request: #{err.class.name}: #{err.message}"
+        Dalli.logger.error err.backtrace.join("\n\t")
       end
 
       include SaslAuthentication
diff --git a/lib/dalli/protocol/binary/request_formatter.rb b/lib/dalli/protocol/binary/request_formatter.rb
index f167ecf2..512590f2 100644
--- a/lib/dalli/protocol/binary/request_formatter.rb
+++ b/lib/dalli/protocol/binary/request_formatter.rb
@@ -31,11 +31,14 @@ class RequestFormatter
           deleteq: 0x14,
           incrq: 0x15,
           decrq: 0x16,
+          flushq: 0x18,
+          appendq: 0x19,
+          prependq: 0x1A,
+          touch: 0x1C,
+          gat: 0x1D,
           auth_negotiation: 0x20,
           auth_request: 0x21,
-          auth_continue: 0x22,
-          touch: 0x1C,
-          gat: 0x1D
+          auth_continue: 0x22
         }.freeze
 
         REQ_HEADER_FORMAT = 'CCnCCnNNQ'
@@ -56,6 +59,8 @@ class RequestFormatter
 
           append: KEY_AND_VALUE,
           prepend: KEY_AND_VALUE,
+          appendq: KEY_AND_VALUE,
+          prependq: KEY_AND_VALUE,
           auth_request: KEY_AND_VALUE,
           auth_continue: KEY_AND_VALUE,
 
@@ -68,8 +73,11 @@ class RequestFormatter
 
           incr: INCR_DECR,
           decr: INCR_DECR,
+          incrq: INCR_DECR,
+          decrq: INCR_DECR,
 
           flush: TTL_ONLY,
+          flushq: TTL_ONLY,
 
           noop: NO_BODY,
           auth_negotiation: NO_BODY,
diff --git a/lib/dalli/protocol/binary/response_processor.rb b/lib/dalli/protocol/binary/response_processor.rb
index 7ebb4dac..49dcb937 100644
--- a/lib/dalli/protocol/binary/response_processor.rb
+++ b/lib/dalli/protocol/binary/response_processor.rb
@@ -58,7 +58,7 @@ def read_header
           read(ResponseHeader::SIZE) || raise(Dalli::NetworkError, 'No response')
         end
 
-        def raise_on_not_ok_status!(resp_header)
+        def raise_on_not_ok!(resp_header)
           return if resp_header.ok?
 
           raise Dalli::DalliError, "Response error #{resp_header.status}: #{RESPONSE_CODES[resp_header.status]}"
@@ -67,24 +67,45 @@ def raise_on_not_ok_status!(resp_header)
         def generic_response(unpack: false, cache_nils: false)
           resp_header, body = read_response
 
-          return cache_nils ? ::Dalli::NOT_FOUND : nil if resp_header.not_found?
           return false if resp_header.not_stored? # Not stored, normal status for add operation
+          return cache_nils ? ::Dalli::NOT_FOUND : nil if resp_header.not_found?
 
-          raise_on_not_ok_status!(resp_header)
+          raise_on_not_ok!(resp_header)
           return true unless body
 
           unpack_response_body(resp_header.extra_len, resp_header.key_len, body, unpack).last
         end
 
-        def data_cas_response
+        ##
+        # Response for a storage operation.  Returns the cas on success.  False
+        # if the value wasn't stored.  And raises an error on all other error
+        # codes from memcached.
+        ##
+        def storage_response
+          resp_header, = read_response
+          return false if resp_header.not_stored? # Not stored, normal status for add operation
+
+          raise_on_not_ok!(resp_header)
+          resp_header.cas
+        end
+
+        def no_body_response
+          resp_header, = read_response
+          return false if resp_header.not_stored? # Not stored, possible status for append/prepend
+
+          raise_on_not_ok!(resp_header)
+          true
+        end
+
+        def data_cas_response(unpack: true)
           resp_header, body = read_response
           return [nil, resp_header.cas] if resp_header.not_found?
           return [nil, false] if resp_header.not_stored?
 
-          raise_on_not_ok_status!(resp_header)
+          raise_on_not_ok!(resp_header)
           return [nil, resp_header.cas] unless body
 
-          [unpack_response_body(resp_header.extra_len, resp_header.key_len, body, true).last, resp_header.cas]
+          [unpack_response_body(resp_header.extra_len, resp_header.key_len, body, unpack).last, resp_header.cas]
         end
 
         def cas_response
@@ -146,31 +167,32 @@ def response_header_from_buffer(buf)
         # complete response is found in the buffer, this will
         # be the response size.  Otherwise it is zero.
         #
-        # The remaining four values in the array are the status, key,
-        # value, and cas returned from the response.
+        # The remaining three values in the array are the ResponseHeader,
+        # key, and value.
         ##
         def getk_response_from_buffer(buf)
           # There's no header in the buffer, so don't advance
-          return [0, 0, nil, nil, nil] unless contains_header?(buf)
+          return [0, nil, nil, nil] unless contains_header?(buf)
 
           resp_header = response_header_from_buffer(buf)
           body_len = resp_header.body_len
 
-          # The response has no body - so we need to advance the
-          # buffer.  This is either the response to the terminating
+          # We have a complete response that has no body.
+          # This is either the response to the terminating
           # noop or, if the status is not zero, an intermediate
           # error response that needs to be discarded.
-          return [ResponseHeader::SIZE, resp_header.status, nil, nil, resp_header.cas] if body_len.zero?
+          return [ResponseHeader::SIZE, resp_header, nil, nil] if body_len.zero?
 
-          # The header is in the buffer, but the body is not
           resp_size = ResponseHeader::SIZE + body_len
-          return [0, resp_header.status, nil, nil, nil] unless buf.bytesize >= resp_size
+          # The header is in the buffer, but the body is not.  As we don't have
+          # a complete response, don't advance the buffer
+          return [0, nil, nil, nil] unless buf.bytesize >= resp_size
 
           # The full response is in our buffer, so parse it and return
           # the values
           body = buf.slice(ResponseHeader::SIZE, body_len)
           key, value = unpack_response_body(resp_header.extra_len, resp_header.key_len, body, true)
-          [resp_size, resp_header.status, key, value, resp_header.cas]
+          [resp_size, resp_header, key, value]
         end
       end
     end
diff --git a/lib/dalli/protocol/connection_manager.rb b/lib/dalli/protocol/connection_manager.rb
new file mode 100644
index 00000000..71b1b173
--- /dev/null
+++ b/lib/dalli/protocol/connection_manager.rb
@@ -0,0 +1,242 @@
+# frozen_string_literal: true
+
+require 'English'
+require 'socket'
+require 'timeout'
+
+module Dalli
+  module Protocol
+    ##
+    # Manages the socket connection to the server, including ensuring liveness
+    # and retries.
+    ##
+    class ConnectionManager
+      DEFAULTS = {
+        # seconds between trying to contact a remote server
+        down_retry_delay: 30,
+        # connect/read/write timeout for socket operations
+        socket_timeout: 1,
+        # times a socket operation may fail before considering the server dead
+        socket_max_failures: 2,
+        # amount of time to sleep between retries when a failure occurs
+        socket_failure_delay: 0.1,
+        # Set keepalive
+        keepalive: true
+      }.freeze
+
+      attr_accessor :hostname, :port, :socket_type, :options
+      attr_reader :sock
+
+      def initialize(hostname, port, socket_type, client_options)
+        @hostname = hostname
+        @port = port
+        @socket_type = socket_type
+        @options = DEFAULTS.merge(client_options)
+        @request_in_progress = false
+        @sock = nil
+        @pid = nil
+
+        reset_down_info
+      end
+
+      def name
+        if socket_type == :unix
+          hostname
+        else
+          "#{hostname}:#{port}"
+        end
+      end
+
+      def establish_connection
+        Dalli.logger.debug { "Dalli::Server#connect #{name}" }
+
+        @sock = memcached_socket
+        @pid = Process.pid
+      rescue SystemCallError, Timeout::Error, EOFError, SocketError => e
+        # SocketError = DNS resolution failure
+        error_on_request!(e)
+      end
+
+      def reconnect_down_server?
+        return true unless @last_down_at
+
+        time_to_next_reconnect = @last_down_at + options[:down_retry_delay] - Time.now
+        return true unless time_to_next_reconnect.positive?
+
+        Dalli.logger.debug do
+          format('down_retry_delay not reached for %<name>s (%<time>.3f seconds left)', name: name,
+                                                                                        time: time_to_next_reconnect)
+        end
+        false
+      end
+
+      def up!
+        log_up_detected
+        reset_down_info
+      end
+
+      # Marks the server instance as down.  Updates the down_at state
+      # and raises an Dalli::NetworkError that includes the underlying
+      # error in the message.  Calls close to clean up socket state
+      def down!
+        close
+        log_down_detected
+
+        @error = $ERROR_INFO&.class&.name
+        @msg ||= $ERROR_INFO&.message
+        raise_down_error
+      end
+
+      def raise_down_error
+        raise Dalli::NetworkError, "#{name} is down: #{@error} #{@msg}"
+      end
+
+      def socket_timeout
+        @socket_timeout ||= @options[:socket_timeout]
+      end
+
+      def confirm_ready!
+        error_on_request!(RuntimeError.new('Already writing to socket')) if request_in_progress?
+        close_on_fork if fork_detected?
+      end
+
+      def close
+        return unless @sock
+
+        begin
+          @sock.close
+        rescue StandardError
+          nil
+        end
+        @sock = nil
+        @pid = nil
+        abort_request!
+      end
+
+      def connected?
+        !@sock.nil?
+      end
+
+      def request_in_progress?
+        @request_in_progress
+      end
+
+      def start_request!
+        @request_in_progress = true
+      end
+
+      def finish_request!
+        @request_in_progress = false
+      end
+
+      def abort_request!
+        @request_in_progress = false
+      end
+
+      def read(count)
+        start_request!
+        data = @sock.readfull(count)
+        finish_request!
+        data
+      rescue SystemCallError, Timeout::Error, EOFError => e
+        error_on_request!(e)
+      end
+
+      def write(bytes)
+        start_request!
+        result = @sock.write(bytes)
+        finish_request!
+        result
+      rescue SystemCallError, Timeout::Error => e
+        error_on_request!(e)
+      end
+
+      # Non-blocking read.  Should only be used in the context
+      # of a caller who has called start_request!, but not yet
+      # called finish_request!.  Here to support the operation
+      # of the get_multi operation
+      def read_nonblock
+        @sock.read_available
+      end
+
+      def max_allowed_failures
+        @max_allowed_failures ||= @options[:socket_max_failures] || 2
+      end
+
+      def error_on_request!(err_or_string)
+        log_warn_message(err_or_string)
+
+        @fail_count += 1
+        if @fail_count >= max_allowed_failures
+          down!
+        else
+          # Closes the existing socket, setting up for a reconnect
+          # on next request
+          reconnect!('Socket operation failed, retrying...')
+        end
+      end
+
+      def reconnect!(message)
+        close
+        sleep(options[:socket_failure_delay]) if options[:socket_failure_delay]
+        raise Dalli::NetworkError, message
+      end
+
+      def reset_down_info
+        @fail_count = 0
+        @down_at = nil
+        @last_down_at = nil
+        @msg = nil
+        @error = nil
+      end
+
+      def memcached_socket
+        if socket_type == :unix
+          Dalli::Socket::UNIX.open(hostname, options)
+        else
+          Dalli::Socket::TCP.open(hostname, port, options)
+        end
+      end
+
+      def log_warn_message(err_or_string)
+        detail = err_or_string.is_a?(String) ? err_or_string : "#{err_or_string.class}: #{err_or_string.message}"
+        Dalli.logger.warn do
+          detail = err_or_string.is_a?(String) ? err_or_string : "#{err_or_string.class}: #{err_or_string.message}"
+          "#{name} failed (count: #{@fail_count}) #{detail}"
+        end
+      end
+
+      def close_on_fork
+        message = 'Fork detected, re-connecting child process...'
+        Dalli.logger.info { message }
+        # Close socket on a fork, setting us up for reconnect
+        # on next request.
+        close
+        raise Dalli::NetworkError, message
+      end
+
+      def fork_detected?
+        @pid && @pid != Process.pid
+      end
+
+      def log_down_detected
+        @last_down_at = Time.now
+
+        if @down_at
+          time = Time.now - @down_at
+          Dalli.logger.debug { format('%<name>s is still down (for %<time>.3f seconds now)', name: name, time: time) }
+        else
+          @down_at = @last_down_at
+          Dalli.logger.warn("#{name} is down")
+        end
+      end
+
+      def log_up_detected
+        return unless @down_at
+
+        time = Time.now - @down_at
+        Dalli.logger.warn { format('%<name>s is back (downtime was %<time>.3f seconds)', name: name, time: time) }
+      end
+    end
+  end
+end
diff --git a/lib/dalli/protocol/response_buffer.rb b/lib/dalli/protocol/response_buffer.rb
index ef3d0b2b..8c127962 100644
--- a/lib/dalli/protocol/response_buffer.rb
+++ b/lib/dalli/protocol/response_buffer.rb
@@ -20,18 +20,21 @@ def read
 
       # Attempts to process a single response from the buffer.  Starts
       # by advancing the buffer to the specified start position
-      def process_single_response(start_position = 0)
-        advance(start_position)
-        @response_processor.getk_response_from_buffer(@buffer)
+      def process_single_getk_response
+        bytes, resp_header, key, value = @response_processor.getk_response_from_buffer(@buffer)
+        advance(bytes)
+        [resp_header, key, value]
       end
 
       # Advances the internal response buffer by bytes_to_advance
       # bytes.  The
       def advance(bytes_to_advance)
+        return unless bytes_to_advance.positive?
+
         @buffer = @buffer[bytes_to_advance..-1]
       end
 
-      # Resets the internal  buffer to an empty state,
+      # Resets the internal buffer to an empty state,
       # so that we're ready to read pipelined responses
       def reset
         @buffer = +''
@@ -42,8 +45,8 @@ def clear
         @buffer = nil
       end
 
-      def completed?
-        @buffer.nil?
+      def in_progress?
+        !@buffer.nil?
       end
     end
   end
diff --git a/lib/dalli/protocol/server_config_parser.rb b/lib/dalli/protocol/server_config_parser.rb
index 455e1f22..8cbda4a3 100644
--- a/lib/dalli/protocol/server_config_parser.rb
+++ b/lib/dalli/protocol/server_config_parser.rb
@@ -19,22 +19,22 @@ class ServerConfigParser
       DEFAULT_PORT = 11_211
       DEFAULT_WEIGHT = 1
 
-      def self.parse(str, client_options)
-        return parse_non_uri(str, client_options) unless str.start_with?(MEMCACHED_URI_PROTOCOL)
+      def self.parse(str)
+        return parse_non_uri(str) unless str.start_with?(MEMCACHED_URI_PROTOCOL)
 
-        parse_uri(str, client_options)
+        parse_uri(str)
       end
 
-      def self.parse_uri(str, client_options)
+      def self.parse_uri(str)
         uri = URI.parse(str)
         auth_details = {
           username: uri.user,
           password: uri.password
         }
-        [uri.host, normalize_port(uri.port), DEFAULT_WEIGHT, :tcp, client_options.merge(auth_details)]
+        [uri.host, normalize_port(uri.port), :tcp, DEFAULT_WEIGHT, auth_details]
       end
 
-      def self.parse_non_uri(str, client_options)
+      def self.parse_non_uri(str)
         res = deconstruct_string(str)
 
         hostname = normalize_host_from_match(str, res)
@@ -45,7 +45,7 @@ def self.parse_non_uri(str, client_options)
           socket_type = :tcp
           port, weight = attributes_for_tcp_socket(res)
         end
-        [hostname, port, weight, socket_type, client_options]
+        [hostname, port, socket_type, weight, {}]
       end
 
       def self.deconstruct_string(str)
diff --git a/test/protocol/test_server_config_parser.rb b/test/protocol/test_server_config_parser.rb
index 8afeeb73..948c0df9 100644
--- a/test/protocol/test_server_config_parser.rb
+++ b/test/protocol/test_server_config_parser.rb
@@ -8,24 +8,22 @@
     let(:weight) { rand(1..5) }
 
     describe 'when the string is not an memcached URI' do
-      let(:options) { {} }
-
       describe 'tcp' do
         describe 'when the hostname is a domain name' do
           let(:hostname) { "a#{SecureRandom.hex(5)}.b#{SecureRandom.hex(3)}.#{%w[net com edu].sample}" }
 
           it 'parses a hostname by itself' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse(hostname, options), [hostname, 11_211, 1, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse(hostname), [hostname, 11_211, :tcp, 1, {}]
           end
 
           it 'parses hostname with a port' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}", options),
-                         [hostname, port, 1, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}"),
+                         [hostname, port, :tcp, 1, {}]
           end
 
           it 'parses hostname with a port and weight' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}:#{weight}", options),
-                         [hostname, port, weight, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}:#{weight}"),
+                         [hostname, port, :tcp, weight, {}]
           end
         end
 
@@ -33,17 +31,17 @@
           let(:hostname) { '203.0.113.28' }
 
           it 'parses a hostname by itself' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse(hostname, options), [hostname, 11_211, 1, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse(hostname), [hostname, 11_211, :tcp, 1, {}]
           end
 
           it 'parses hostname with a port' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}", options),
-                         [hostname, port, 1, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}"),
+                         [hostname, port, :tcp, 1, {}]
           end
 
           it 'parses hostname with a port and weight' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}:#{weight}", options),
-                         [hostname, port, weight, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}:#{weight}"),
+                         [hostname, port, :tcp, weight, {}]
           end
         end
 
@@ -51,18 +49,18 @@
           let(:hostname) { ['2001:db8:ffff:ffff:ffff:ffff:ffff:ffff', '2001:db8::'].sample }
 
           it 'parses a hostname by itself' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("[#{hostname}]", options),
-                         [hostname, 11_211, 1, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("[#{hostname}]"),
+                         [hostname, 11_211, :tcp, 1, {}]
           end
 
           it 'parses hostname with a port' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("[#{hostname}]:#{port}", options),
-                         [hostname, port, 1, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("[#{hostname}]:#{port}"),
+                         [hostname, port, :tcp, 1, {}]
           end
 
           it 'parses hostname with a port and weight' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse("[#{hostname}]:#{port}:#{weight}", options),
-                         [hostname, port, weight, :tcp, {}]
+            assert_equal Dalli::Protocol::ServerConfigParser.parse("[#{hostname}]:#{port}:#{weight}"),
+                         [hostname, port, :tcp, weight, {}]
           end
         end
       end
@@ -71,17 +69,17 @@
         let(:hostname) { "/tmp/#{SecureRandom.hex(5)}" }
 
         it 'parses a socket by itself' do
-          assert_equal Dalli::Protocol::ServerConfigParser.parse(hostname, {}), [hostname, nil, 1, :unix, {}]
+          assert_equal Dalli::Protocol::ServerConfigParser.parse(hostname), [hostname, nil, :unix, 1, {}]
         end
 
         it 'parses socket with a weight' do
-          assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{weight}", {}),
-                       [hostname, nil, weight, :unix, {}]
+          assert_equal Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{weight}"),
+                       [hostname, nil, :unix, weight, {}]
         end
 
         it 'produces an error with a port and weight' do
           err = assert_raises Dalli::DalliError do
-            Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}:#{weight}", {})
+            Dalli::Protocol::ServerConfigParser.parse("#{hostname}:#{port}:#{weight}")
           end
           assert_equal err.message, "Could not parse hostname #{hostname}:#{port}:#{weight}"
         end
@@ -97,37 +95,18 @@
       describe 'when the URI is properly formed and includes all values' do
         let(:uri) { "memcached://#{user}:#{password}@#{hostname}:#{port}" }
 
-        describe 'when the client options are empty' do
-          let(:client_options) { {} }
-
-          it 'parses correctly' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse(uri, client_options),
-                         [hostname, port, 1, :tcp, { username: user, password: password }]
-          end
-        end
-
-        describe 'when the client options are not empty' do
-          let(:option_a) { SecureRandom.hex(3) }
-          let(:option_b) { SecureRandom.hex(3) }
-          let(:client_options) { { a: option_a, b: option_b } }
-
-          it 'parses correctly' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse(uri, client_options),
-                         [hostname, port, 1, :tcp, { username: user, password: password, a: option_a, b: option_b }]
-          end
+        it 'parses correctly' do
+          assert_equal Dalli::Protocol::ServerConfigParser.parse(uri),
+                       [hostname, port, :tcp, 1, { username: user, password: password }]
         end
       end
 
       describe 'when the URI does not include a port' do
         let(:uri) { "memcached://#{user}:#{password}@#{hostname}" }
 
-        describe 'when the client options are empty' do
-          let(:client_options) { {} }
-
-          it 'parses correctly' do
-            assert_equal Dalli::Protocol::ServerConfigParser.parse(uri, client_options),
-                         [hostname, 11_211, 1, :tcp, { username: user, password: password }]
-          end
+        it 'parses correctly' do
+          assert_equal Dalli::Protocol::ServerConfigParser.parse(uri),
+                       [hostname, 11_211, :tcp, 1, { username: user, password: password }]
         end
       end
     end
@@ -136,7 +115,7 @@
       describe 'when the string is empty' do
         it 'produces an error' do
           err = assert_raises Dalli::DalliError do
-            Dalli::Protocol::ServerConfigParser.parse('', {})
+            Dalli::Protocol::ServerConfigParser.parse('')
           end
           assert_equal('Could not parse hostname ', err.message)
         end
@@ -145,7 +124,7 @@
       describe 'when the string starts with a colon' do
         it 'produces an error' do
           err = assert_raises Dalli::DalliError do
-            Dalli::Protocol::ServerConfigParser.parse(':1:2', {})
+            Dalli::Protocol::ServerConfigParser.parse(':1:2')
           end
           assert_equal('Could not parse hostname :1:2', err.message)
         end
@@ -154,7 +133,7 @@
       describe 'when the string ends with a colon' do
         it 'produces an error' do
           err = assert_raises Dalli::DalliError do
-            Dalli::Protocol::ServerConfigParser.parse('abc.com:', {})
+            Dalli::Protocol::ServerConfigParser.parse('abc.com:')
           end
           assert_equal('Could not parse hostname abc.com:', err.message)
         end
diff --git a/test/test_binary_protocol.rb b/test/test_binary_protocol.rb
index 29164e5d..c13a47c0 100644
--- a/test/test_binary_protocol.rb
+++ b/test/test_binary_protocol.rb
@@ -99,22 +99,22 @@
     end
   end
 
-  describe 'process_outstanding_pipeline_requests' do
+  describe 'pipeline_next_responses' do
     subject { Dalli::Protocol::Binary.new('127.0.0.1') }
 
-    it 'raises NetworkError when called before pipeline_response_start' do
+    it 'raises NetworkError when called before pipeline_response_setup' do
       assert_raises Dalli::NetworkError do
         subject.request(:pipelined_get, %w[a b])
-        subject.process_outstanding_pipeline_requests
+        subject.pipeline_next_responses
       end
     end
 
-    it 'raises NetworkError when called after pipeline_response_abort' do
+    it 'raises NetworkError when called after pipeline_abort' do
       assert_raises Dalli::NetworkError do
         subject.request(:pipelined_get, %w[a b])
-        subject.pipeline_response_start
-        subject.pipeline_response_abort
-        subject.process_outstanding_pipeline_requests
+        subject.pipeline_response_setup
+        subject.pipeline_abort
+        subject.pipeline_next_responses
       end
     end
   end
diff --git a/test/test_dalli.rb b/test/test_dalli.rb
index 2b7fc082..6e31748b 100644
--- a/test/test_dalli.rb
+++ b/test/test_dalli.rb
@@ -310,7 +310,7 @@
       end
     end
 
-    it 'supports get multi' do
+    it 'supports pipelined get' do
       memcached_persistent do |dc|
         dc.close
         dc.flush
@@ -332,7 +332,7 @@
         end
         assert_empty expected_resp
 
-        # Perform a big multi-get with 1000 elements.
+        # Perform a big quiet set with 1000 elements.
         arr = []
         dc.multi do
           1000.times do |idx|
@@ -341,52 +341,13 @@
           end
         end
 
+        # Retrieve the elements with a pipelined get
         result = dc.get_multi(arr)
         assert_equal(1000, result.size)
         assert_equal(50, result['50'])
       end
     end
 
-    it 'does not corrupt multi blocks with errors' do
-      memcached_persistent do |dc|
-        dc.close
-        dc.flush
-        dc.set('a', 'av')
-        dc.set('b', 'bv')
-        assert_equal 'av', dc.get('a')
-        assert_equal 'bv', dc.get('b')
-
-        refute Thread.current[::Dalli::MULTI_KEY]
-        dc.multi do
-          assert Thread.current[::Dalli::MULTI_KEY]
-          dc.delete('non_existent_key')
-        end
-        refute Thread.current[::Dalli::MULTI_KEY]
-        assert_equal 'av', dc.get('a')
-        assert_equal 'bv', dc.get('b')
-      end
-    end
-
-    it 'raises an error if an invalid operation is used in a multi block' do
-      memcached_persistent do |dc|
-        dc.close
-        dc.flush
-        dc.set('a', 'av')
-        dc.set('b', 'bv')
-        assert_equal 'av', dc.get('a')
-        assert_equal 'bv', dc.get('b')
-
-        refute Thread.current[::Dalli::MULTI_KEY]
-        dc.multi do
-          assert Thread.current[::Dalli::MULTI_KEY]
-          assert_raises Dalli::NotPermittedMultiOpError do
-            dc.get('a')
-          end
-        end
-        refute Thread.current[::Dalli::MULTI_KEY]
-      end
-    end
-
     it 'support raw incr/decr' do
       memcached_persistent do |client|
         client.flush
@@ -532,7 +493,7 @@
         dc.set(:a, 1)
         ring = dc.send(:ring)
         server = ring.servers.first
-        socket = server.instance_variable_get('@sock')
+        socket = server.sock
 
         optval = socket.getsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE)
         optval = optval.unpack 'i'
diff --git a/test/test_quiet.rb b/test/test_quiet.rb
new file mode 100644
index 00000000..b7597ab5
--- /dev/null
+++ b/test/test_quiet.rb
@@ -0,0 +1,239 @@
+# frozen_string_literal: true
+
+require_relative 'helper'
+
+describe 'Dalli' do
+  it 'supports the use of set in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      value = SecureRandom.hex(3)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.set(key, value)
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_equal value, dc.get(key)
+    end
+  end
+
+  it 'supports the use of add in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      existing = SecureRandom.hex(4)
+      oldvalue = SecureRandom.hex(3)
+      value = SecureRandom.hex(3)
+      dc.set(existing, oldvalue)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.add(key, value)
+
+        # Should handle error case without error or unexpected behavior
+        assert_nil dc.add(existing, value)
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_equal value, dc.get(key)
+      assert_equal oldvalue, dc.get(existing)
+    end
+  end
+
+  it 'supports the use of replace in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      nonexistent = SecureRandom.hex(4)
+      oldvalue = SecureRandom.hex(3)
+      value = SecureRandom.hex(3)
+      dc.set(key, oldvalue)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.replace(key, value)
+
+        # Should handle error case without error or unexpected behavior
+        assert_nil dc.replace(nonexistent, value)
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_equal value, dc.get(key)
+      assert_nil dc.get(nonexistent)
+    end
+  end
+
+  it 'supports the use of delete in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      existing = SecureRandom.hex(4)
+      value = SecureRandom.hex(3)
+      dc.set(existing, value)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.delete(existing)
+
+        # Should handle error case without error or unexpected behavior
+        assert_nil dc.delete(key)
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_nil dc.get(existing)
+      assert_nil dc.get(key)
+    end
+  end
+
+  it 'supports the use of append in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      value = SecureRandom.hex(3)
+      dc.set(key, value, 90, raw: true)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.append(key, 'abc')
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_equal "#{value}abc", dc.get(key)
+    end
+  end
+
+  it 'supports the use of prepend in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      value = SecureRandom.hex(3)
+      dc.set(key, value, 90, raw: true)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.prepend(key, 'abc')
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_equal "abc#{value}", dc.get(key)
+    end
+  end
+
+  it 'supports the use of incr in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      value = 546
+      incr = 134
+      dc.set(key, value, 90, raw: true)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.incr(key, incr)
+      end
+      refute Thread.current[::Dalli::QUIET]
+
+      assert_equal 680, dc.get(key).to_i
+    end
+  end
+
+  it 'supports the use of decr in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      key = SecureRandom.hex(3)
+      value = 546
+      incr = 134
+      dc.set(key, value, 90, raw: true)
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be nil
+        assert_nil dc.decr(key, incr)
+      end
+
+      assert_equal 412, dc.get(key).to_i
+    end
+  end
+
+  it 'supports the use of flush in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      refute Thread.current[::Dalli::QUIET]
+      dc.quiet do
+        assert Thread.current[::Dalli::QUIET]
+
+        # Response should be a non-empty array of nils
+        arr = dc.flush(90)
+        assert_equal 2, arr.size
+        assert arr.all?(&:nil?)
+      end
+      refute Thread.current[::Dalli::QUIET]
+    end
+  end
+
+  it 'does not corrupt the underlying response buffer when a memcached error occurs in a quiet block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      dc.set('a', 'av')
+      dc.set('b', 'bv')
+      assert_equal 'av', dc.get('a')
+      assert_equal 'bv', dc.get('b')
+
+      refute Thread.current[::Dalli::QUIET]
+      dc.multi do
+        assert Thread.current[::Dalli::QUIET]
+        dc.delete('non_existent_key')
+      end
+      refute Thread.current[::Dalli::QUIET]
+      assert_equal 'av', dc.get('a')
+      assert_equal 'bv', dc.get('b')
+    end
+  end
+
+  it 'raises an error if an invalid operation is used in a multi block' do
+    memcached_persistent do |dc|
+      dc.close
+      dc.flush
+      dc.set('a', 'av')
+      dc.set('b', 'bv')
+      assert_equal 'av', dc.get('a')
+      assert_equal 'bv', dc.get('b')
+
+      refute Thread.current[::Dalli::QUIET]
+      dc.multi do
+        assert Thread.current[::Dalli::QUIET]
+        assert_raises Dalli::NotPermittedMultiOpError do
+          dc.get('a')
+        end
+      end
+      refute Thread.current[::Dalli::QUIET]
+    end
+  end
+end