func (socket *mongoSocket) readLoop() {
	p := make([]byte, 36) // 16 from header + 20 from OP_REPLY fixed fields
	s := make([]byte, 4)
	conn := socket.conn // No locking, conn never changes.
	for {
		// XXX Handle timeouts, , etc
		err := fill(conn, p)
		if err != nil {
			socket.kill(err, true)
			return
		}

		totalLen := getInt32(p, 0)
		responseTo := getInt32(p, 8)
		opCode := getInt32(p, 12)

		// Don't use socket.server.Addr here.  socket is not
		// locked and socket.server may go away.
		debugf("Socket %p to %s: got reply (%d bytes)", socket, socket.addr, totalLen)

		_ = totalLen

		if opCode != 1 {
			socket.kill(errors.New("opcode != 1, corrupted data?"), true)
			return
		}

		reply := replyOp{
			flags:     uint32(getInt32(p, 16)),
			cursorId:  getInt64(p, 20),
			firstDoc:  getInt32(p, 28),
			replyDocs: getInt32(p, 32),
		}

		stats.receivedOps(+1)
		stats.receivedDocs(int(reply.replyDocs))

		socket.Lock()
		replyFunc, ok := socket.replyFuncs[uint32(responseTo)]
		if ok {
			delete(socket.replyFuncs, uint32(responseTo))
		}
		socket.Unlock()

		if replyFunc != nil && reply.replyDocs == 0 {
			replyFunc(nil, &reply, -1, nil)
		} else {
			for i := 0; i != int(reply.replyDocs); i++ {
				err := fill(conn, s)
				if err != nil {
					if replyFunc != nil {
						replyFunc(err, nil, -1, nil)
					}
					socket.kill(err, true)
					return
				}

				b := make([]byte, int(getInt32(s, 0)))

				// copy(b, s) in an efficient way.
				b[0] = s[0]
				b[1] = s[1]
				b[2] = s[2]
				b[3] = s[3]

				err = fill(conn, b[4:])
				if err != nil {
					if replyFunc != nil {
						replyFunc(err, nil, -1, nil)
					}
					socket.kill(err, true)
					return
				}

				if globalDebug && globalLogger != nil {
					m := bson.M{}
					if err := bson.Unmarshal(b, m); err == nil {
						debugf("Socket %p to %s: received document: %#v", socket, socket.addr, m)
					}
				}

				if replyFunc != nil {
					replyFunc(nil, &reply, i, b)
				}

				// XXX Do bound checking against totalLen.
			}
		}

		socket.Lock()
		if len(socket.replyFuncs) == 0 {
			// Nothing else to read for now. Disable deadline.
			socket.conn.SetReadDeadline(time.Time{})
		} else {
			socket.updateDeadline(readDeadline)
		}
		socket.Unlock()

		// XXX Do bound checking against totalLen.
	}
}
func (socket *mongoSocket) readLoop() {
	p := make([]byte, 36) 
	s := make([]byte, 4)
	conn := socket.conn 
	for {
		
		err := fill(conn, p)
		if err != nil {
			socket.kill(err, true)
			return
		}

		totalLen := getInt32(p, 0)
		responseTo := getInt32(p, 8)
		opCode := getInt32(p, 12)

		
		
		debugf("Socket %p to %s: got reply (%d bytes)", socket, socket.addr, totalLen)

		_ = totalLen

		if opCode != 1 {
			socket.kill(errors.New("opcode != 1, corrupted data?"), true)
			return
		}

		reply := replyOp{
			flags:     uint32(getInt32(p, 16)),
			cursorId:  getInt64(p, 20),
			firstDoc:  getInt32(p, 28),
			replyDocs: getInt32(p, 32),
		}

		stats.receivedOps(+1)
		stats.receivedDocs(int(reply.replyDocs))

		socket.Lock()
		replyFunc, ok := socket.replyFuncs[uint32(responseTo)]
		if ok {
			delete(socket.replyFuncs, uint32(responseTo))
		}
		socket.Unlock()

		if replyFunc != nil && reply.replyDocs == 0 {
			replyFunc(nil, &reply, -1, nil)
		} else {
			for i := 0; i != int(reply.replyDocs); i++ {
				err := fill(conn, s)
				if err != nil {
					if replyFunc != nil {
						replyFunc(err, nil, -1, nil)
					}
					socket.kill(err, true)
					return
				}

				b := make([]byte, int(getInt32(s, 0)))

				
				b[0] = s[0]
				b[1] = s[1]
				b[2] = s[2]
				b[3] = s[3]

				err = fill(conn, b[4:])
				if err != nil {
					if replyFunc != nil {
						replyFunc(err, nil, -1, nil)
					}
					socket.kill(err, true)
					return
				}

				if globalDebug && globalLogger != nil {
					m := bson.M{}
					if err := bson.Unmarshal(b, m); err == nil {
						debugf("Socket %p to %s: received document: %#v", socket, socket.addr, m)
					}
				}

				if replyFunc != nil {
					replyFunc(nil, &reply, i, b)
				}

				
			}
		}

		socket.Lock()
		if len(socket.replyFuncs) == 0 {
			
			socket.conn.SetReadDeadline(time.Time{})
		} else {
			socket.updateDeadline(readDeadline)
		}
		socket.Unlock()

		
	}
}
func ( socket * mongoSocket ) readLoop ( ) { p := make ( [ ] byte , 36 ) // 16 from header + 20 from OP_REPLY fixed fields s := make ( [ ] byte , 4 ) conn := socket . conn // No locking, conn never changes. for { // XXX Handle timeouts, , etc err := fill ( conn , p ) if err != nil { socket . kill ( err , true ) return } totalLen := getInt32 ( p , 0 ) responseTo := getInt32 ( p , 8 ) opCode := getInt32 ( p , 12 ) // Don't use socket.server.Addr here.  socket is not // locked and socket.server may go away. debugf ( " " , socket , socket . addr , totalLen ) _ = totalLen if opCode != 1 { socket . kill ( errors . New ( " " ) , true ) return } reply := replyOp { flags : uint32 ( getInt32 ( p , 16 ) ) , cursorId : getInt64 ( p , 20 ) , firstDoc : getInt32 ( p , 28 ) , replyDocs : getInt32 ( p , 32 ) , } stats . receivedOps ( + 1 ) stats . receivedDocs ( int ( reply . replyDocs ) ) socket . Lock ( ) replyFunc , ok := socket . replyFuncs [ uint32 ( responseTo ) ] if ok { delete ( socket . replyFuncs , uint32 ( responseTo ) ) } socket . Unlock ( ) if replyFunc != nil && reply . replyDocs == 0 { replyFunc ( nil , & reply , - 1 , nil ) } else { for i := 0 ; i != int ( reply . replyDocs ) ; i ++ { err := fill ( conn , s ) if err != nil { if replyFunc != nil { replyFunc ( err , nil , - 1 , nil ) } socket . kill ( err , true ) return } b := make ( [ ] byte , int ( getInt32 ( s , 0 ) ) ) // copy(b, s) in an efficient way. b [ 0 ] = s [ 0 ] b [ 1 ] = s [ 1 ] b [ 2 ] = s [ 2 ] b [ 3 ] = s [ 3 ] err = fill ( conn , b [ 4 : ] ) if err != nil { if replyFunc != nil { replyFunc ( err , nil , - 1 , nil ) } socket . kill ( err , true ) return } if globalDebug && globalLogger != nil { m := bson . M { } if err := bson . Unmarshal ( b , m ) ; err == nil { debugf ( " " , socket , socket . addr , m ) } } if replyFunc != nil { replyFunc ( nil , & reply , i , b ) } // XXX Do bound checking against totalLen. } } socket . Lock ( ) if len ( socket . replyFuncs ) == 0 { // Nothing else to read for now. Disable deadline. socket . conn . SetReadDeadline ( time . Time { } ) } else { socket . updateDeadline ( readDeadline ) } socket . Unlock ( ) // XXX Do bound checking against totalLen. } }
// Estimated minimum cost per socket: 1 goroutine + memory for the largest
// document ever seen.
Estimated minimum cost per socket : 1 goroutine + memory for the largest document ever seen .