«Закрыто явно» mgo MongoDB в Go

Question

У меня есть кластер Докеров, в котором работает 10 веб-сервисов (такого же типа). Все они используют MongoDB, среди прочего, для сохранения данных.

Это код, который вызывается из main(), когда служба загружается:

// Init establishes a connection with MongoDB instance. 
func Init(mongoURL string) *mgo.Session { 
    mongo, err := mgo.Dial(mongoURL) 
    misc.PanicIf(err) 

    // make sure we are strongly consistent 
    mongo.SetMode(mgo.Strong, true) 

    // update global state 
    db = mongo 
    Entries = db.DB("").C("entries") 
    Channels = db.DB("").C("channels") 
    Settings = db.DB("").C("settings") 
    Metadata = db.DB("").C("metadata") 

    // only use this on first load, to confirm the settings are there 
    // every refresh should be done via UpdateGlobalSettings (thread-safe) 
    GlobalSettings = &GlobalSettingsStruct{} 
    GlobalSettings.Init() 

    return mongo 
} 

Поэтому в основном API и рабочие просто использовать глобальные переменные, такие как Entries, Settings и т.д.

После некоторого времени работы служба перестала работать должным образом. Каждое действие mongo, такое как Entries.find(...), возвращает ошибку: Closed Explicitly.

Что это значит? Должен ли я периодически обновлять соединение mongoDB или я должен закрыть его и восстановить соединение с каждым запросом?

Приложение ориентировано на производительность, поэтому, несмотря на то, что соединение mongo не работает, все работает и работает, поскольку все работает в кэше в памяти или кластере. Я не хочу делать что-то глупое, что задерживает обработку.

Answer 1

Прежде всего попробуйте включить режим отладки в mgo, чтобы получить дополнительную информацию о том, что происходит.

Я полагаю, что сервер отключает соединение после некоторого времени бездействия. В любом случае, обычно подход заключается в том, чтобы сделать mgo Dial, а затем Copy соединение в начале обработки каждого запроса (с использованием промежуточного программного обеспечения).

Pietro

Answer 2

Это нормально. но использовать Session.Copy для создания нового экземпляра подключения вместо прямого использования возвращенного сеанса. есть пул соединений внутри пакета драйвера golang mongodb.

Answer 3

Когда sess.SetPoolLimit(50) не используется, многие ошибки возникают, когда mgo находится под напряжением, например, 10.000 одновременных соединений.

Когда я ограничил пул, ошибки ушли.

Я создал исходный код тестового примера для этой проблемы ниже, чтобы вы могли легко протестировать его на своей собственной машине.

У вас есть предложения по поводу этого поведения, я хотел бы услышать.

package main 

import (
    "fmt" 
    "time" 

    // you can also use original go-mgo/mgo here as well 
    mgo "github.com/globalsign/mgo" 
    "github.com/globalsign/mgo/bson" 
) 

// TODO: put some records into db first: 
// 
// use testapi 
// db.competitions.insert([ 
// {game_id: 1, game_name: "foo"}, 
// {game_id: 2, game_name: "bar"}, 
// {game_id: 3, game_name: "jazz"} 
// ]) 

// NOTE: you might want to increase this depending on your machine power 
//  mine is: 
//   MacBook (Retina, 12-inch, Early 2015) 
//   1,2 GHz Intel Core M 
//   8 GB 1600 MHz DDR3 
const ops = 10000 

type m bson.M 

func main() { 
    sess, err := mgo.DialWithTimeout("localhost", time.Second) 
    if err != nil { 
     panic(err) 
    } 
    defer sess.Close() 

    // NOTE: without this setting there are many errors 
    //  see the end of the file 
    // setting pool limit prevents most of the timeouts 
    // sess.SetPoolLimit(50) 

    // sess.SetSocketTimeout(60 * time.Second) 
    sess.SetMode(mgo.Monotonic, true) 

    time.Sleep(time.Second) 

    done := make(chan bool, ops) 

    for i := 0; i < ops; i++ { 
     go func() { 
      defer func() { done <- true }() 

      result, err := fetchSomething(sess) 
      if err != nil { 
       fmt.Printf("ERR: %s\n", err) 
      } 
      fmt.Printf("RESULT: %+v\n", result) 
     }() 
    } 

    for i := 0; i < ops; i++ { 
     <-done 
    } 
} 

func fetchSomething(sess *mgo.Session) ([]m, error) { 
    s := sess.Copy() 
    defer s.Close() 

    result := []m{} 

    group := m{ 
     "$group": m{ 
      "_id": m{ 
       "id": "$game_id", 
       "name": "$game_name", 
      }, 
     }, 
    } 

    project := m{ 
     "$project": m{ 
      "_id": "$_id.id", 
      "name": "$_id.name", 
     }, 
    } 

    sort := m{ 
     "$sort": m{ 
      "_id": 1, 
     }, 
    } 

    err := col(s, "competitions").Pipe([]m{group, project, sort}).All(&result) 
    if err != nil { 
     return nil, err 
    } 
    return result, nil 
} 

// col is a helper for selecting a column 
func col(sess *mgo.Session, name string) *mgo.Collection { 
    return sess.DB("testapi").C(name) 
} 

/* 


ERRORS WITHOUT sess.SetPoolLimit(50) 


$ go run socket_error.go 
RESULT: [map[name:foo _id:1] map[_id:2 name:bar] map[_id:3 name:jazz]] 
ERR: read tcp 127.0.0.1:52918->127.0.0.1:27017: read: connection reset by peer 
ERR: write tcp 127.0.0.1:52084->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
RESULT: [] 
ERR: write tcp 127.0.0.1:53627->127.0.0.1:27017: write: broken pipe 
ERR: write tcp 127.0.0.1:53627->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
ERR: write tcp 127.0.0.1:53627->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
ERR: write tcp 127.0.0.1:53627->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
ERR: write tcp 127.0.0.1:53627->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
ERR: write tcp 127.0.0.1:53627->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
RESULT: [] 
ERR: read tcp 127.0.0.1:53627->127.0.0.1:27017: read: connection reset by peer 
RESULT: [] 
ERR: resetNonce: write tcp 127.0.0.1:53625->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
RESULT: [map[name:foo _id:1] map[_id:2 name:bar] map[_id:3 name:jazz]] 
ERR: resetNonce: write tcp 127.0.0.1:54591->127.0.0.1:27017: write: broken pipe 
RESULT: [] 
... 
... 
*/