// Block the IP
        for (int i = 1; i <= 5; i++) {
            assertTrue(rateLimiter.checkAttempt("user" + i, ip));
            rateLimiter.recordFailure("user" + i, ip);
        }

        // Should be blocked
        assertFalse(rateLimiter.checkAttempt("user6", ip), "IP should be blocked");

        // Manually unblock
        assertTrue(rateLimiter.unblockIp(ip), "Manual unblock should succeed");

		return
	}
	if m.acked || m.outBlock == nil {
		return
	}
	available := cap(m.outBlock)
	for range available {
		m.outBlock <- struct{}{}
	}
	m.acked = true
}

func (m *muxClient) unblockSend(seq uint32) {
	if !m.checkSeq(seq) {
		return
	}
	select {
	case m.outBlock <- struct{}{}:
	default:
		gridLogIf(m.ctx, errors.New("output unblocked overflow"))
	}
}

		if !ok || mid == nil {
			resp(true)
			return
		}
		mid.respMu.Lock()
		resp(mid.closed)
		mid.respMu.Unlock()
	case debugBlockInboundMessages:
		c.connMu.Lock()
		a, _ := args[0].(chan struct{})
		block := (<-chan struct{})(a)
		c.blockMessages.Store(&block)
		c.connMu.Unlock()
	}
}

// wsWriter writes websocket messages.
type wsWriter struct {
	tmp [ws.MaxHeaderSize]byte
}

    connection.setReadTimeout(500)
    try {
      connection.getResponseCode()
      fail<Any>()
    } catch (expected: SocketTimeoutException) {
    }

    // Shutting down the server should unblock the dispatcher.
    server.shutdown()
  }

  @Test
  fun requestUrlReconstructed() {
    server.enqueue(MockResponse().setBody("hello world"))
    val url = server.url("/a/deep/path?key=foo%20bar").toUrl()

    connection.readTimeout = 500
    assertFailsWith<SocketTimeoutException> {
      connection.responseCode
    }

    // Closing the server should unblock the dispatcher.
    server.close()
  }

  @Test
  fun requestUrlReconstructed() {
    server.enqueue(
      MockResponse
        .Builder()
        .body("hello world")
        .build(),

	// Caller may use this as an indication to call again.
	return !checkFailedUnlocks(*locks, tolerance)
}

// Unlock unlocks the write lock.
//
// It is a run-time error if dm is not locked on entry to Unlock.
func (dm *DRWMutex) Unlock(ctx context.Context) {
	dm.m.Lock()
	dm.cancelRefresh()
	dm.m.Unlock()

	restClnts, owner := dm.clnt.GetLockers()
	// create temp array on stack
	locks := make([]string, len(restClnts))

		globalHTTPStats.incS3RequestsIncoming()

		if r.Header.Get(globalObjectPerfUserMetadata) == "" {
			if val := globalServiceFreeze.Load(); val != nil {
				if unlock, ok := val.(chan struct{}); ok && unlock != nil {
					// Wait until unfrozen.
					select {
					case <-unlock:
					case <-r.Context().Done():
						// if client canceled we don't need to wait here forever.
						return
					}
				}
			}
		}

			select {
			// Check after a second
			case <-time.After(time.Second):
				h.mutex.Unlock()
				continue

			case <-h.ctx.Done():
				h.mutex.Unlock()
				// discard result and return.
				return errHealStopSignalled

			// Timeout if no results consumed for too long.
			case <-unconsumedTimer.C:
				h.mutex.Unlock()
				return errHealIdleTimeout
			}
		}
		break
	}

	if stats == nil {
		return
	}
	stats.Lock()
	defer stats.Unlock()
	if stats.apiStats == nil {
		stats.apiStats = make(map[string]int)
	}
	stats.apiStats[api]++
}

// Dec increments the api stats counter.
func (stats *HTTPAPIStats) Dec(api string) {
	if stats == nil {
		return
	}
	stats.Lock()
	defer stats.Unlock()
	if val, ok := stats.apiStats[api]; ok && val > 0 {
		stats.apiStats[api]--

	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()
	delete(r.Cache, bucket)
}

// UpdateReplicaStat updates in-memory replica statistics with new values.
func (r *ReplicationStats) UpdateReplicaStat(bucket string, n int64) {
	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()
	bs, ok := r.Cache[bucket]
	if !ok {
		bs = newBucketReplicationStats()
	}