 		}
 
 		let bufferingObserver: Signal<Value, Error>.Observer = Observer { event in
-			// Send serially with respect to other senders, and never while
-			// another thread is in the process of replaying.
-			dispatch_sync(queue) {
-				let originalState = state.modify { state in
-					var mutableState = state
-
-					if let value = event.value {
-						mutableState.addValue(value, upToCapacity: capacity)
-					} else {
-						// Disconnect all observers and prevent future
-						// attachments.
-						mutableState.terminationEvent = event
-						mutableState.observers = nil
-					}
-
-					return mutableState
-				}
-
-				if let observers = originalState.observers {
-					for observer in observers {
-						observer.action(event)
-					}
+			let originalState = state.modify { state in
+				var mutableState = state
+				
+				if let value = event.value {
+					mutableState.addValue(value, upToCapacity: capacity)
+				} else {
+					// Disconnect all observers and prevent future
+					// attachments.
+					mutableState.terminationEvent = event
+					mutableState.observers = nil
+				}
+				
+				return mutableState
+			}
+			
+			if let observers = originalState.observers {
+				for observer in observers {
+					observer.action(event)
 				}
 			}
 		}

	public static func buffer(capacity: Int) -> (SignalProducer, Signal<Value, Error>.Observer) {
		precondition(capacity >= 0, "Invalid capacity: \(capacity)")

		// This is effectively used as a synchronous mutex, but permitting
		// limited recursive locking (see below).
		//
		// The queue is a "variable" just so we can use its address as the key
		// and the value for dispatch_queue_set_specific().
		var queue = dispatch_queue_create("org.reactivecocoa.ReactiveCocoa.SignalProducer.buffer", DISPATCH_QUEUE_SERIAL)
		dispatch_queue_set_specific(queue, &queue, &queue, nil)

		// Used as an atomic variable so we can remove observers without needing
		// to run on the queue.
		let state: Atomic<BufferState<Value, Error>> = Atomic(BufferState())

	/// The observers currently attached to the buffered producer, or nil if the
	/// producer was terminated.
	var observers: Bag<Signal<Value, Error>.Observer>? = Bag()
	
	/// The set of unused replay token identifiers.
	var replayBuffers: Bag<ReplayBuffer<Value>>? = Bag()

	/// Appends a new value to the buffer, trimming it down to the given capacity
	/// if necessary.
	mutating func addValue(value: Value, upToCapacity capacity: Int) {
		precondition(capacity >= 0)
		
		if let replayBuffers = replayBuffers {
			for buffer in replayBuffers {
				buffer.values.append(value)
			}
		}

		if capacity == 0 {
			values = []
