Add configurable MaxRequestsPerConn cluster param

Before this change, a connection had a hardcoded maximum number of
streams. Since this value was so large (32768 for CQL v3+) this
essentially meant an unbound concurrency.

This commit allows a user to configure this parameter by a newly
added MaxRequestsPerConn option.

The "MaxRequestsPerConn" naming was chosen to be very general
deliberately to allow us to safely introduce a more advanced behavior in
the future: where the concurrency limiting happens not at a IDGenerator
level and an additional client-side queue is introduced for requests
to wait to be sent to Scylla (in addition to inflight requests).

This commit deliberately does not change the defaults as to be
non-pessimizing for the users.

Fixes #112

cluster.go

@@ -72,6 +72,11 @@ type ClusterConfig struct {
 	// Default: 2
 	NumConns int
 
+	// Maximum number of inflight requests allowed per connection.
+	// Default: 32768 for CQL v3 and newer
+	// Default: 128 for older CQL versions
+	MaxRequestsPerConn int
+
 	// Default consistency level.
 	// Default: Quorum
 	Consistency Consistency

conn.go

@@ -287,7 +287,7 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 		errorHandler:  errorHandler,
 		compressor:    cfg.Compressor,
 		session:       s,
-		streams:       streams.New(cfg.ProtoVersion),
+		streams:       s.streamIDGenerator(cfg.ProtoVersion),
 		host:          host,
 		isSchemaV2:    true, // Try using "system.peers_v2" until proven otherwise
 		frameObserver: s.frameObserver,
@@ -313,6 +313,13 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 	return c, nil
 }
 
+func (s *Session) streamIDGenerator(protocol int) *streams.IDGenerator {
+	if s.cfg.MaxRequestsPerConn > 0 {
+		return streams.NewLimited(s.cfg.MaxRequestsPerConn)
+	}
+	return streams.New(protocol)
+}
+
 func (c *Conn) init(ctx context.Context, dialedHost *DialedHost) error {
 	if c.session.cfg.AuthProvider != nil {
 		var err error

internal/streams/streams.go

@@ -24,7 +24,10 @@ func New(protocol int) *IDGenerator {
 	if protocol > 2 {
 		maxStreams = 32768
 	}
+	return NewLimited(maxStreams)
+}
 
+func NewLimited(maxStreams int) *IDGenerator {
 	buckets := maxStreams / 64
 	// reserve stream 0
 	streams := make([]uint64, buckets)