Getting "An open statement already exists" while data is correct

[oharlem]

Hi,

Thanks for working on the bolt driver!

I observe the following issue: the following query will return correct result if run with relatively large intervals. However, if interval between queries decrease to a certain level, the driver throws "Error #01: An open statement already exists".

How to reproduce:

1. Works fine if, for example, you'll have an endpoint/script that you'll be triggering manually. I could not reach low enough interval to catch the error.
2. However, if you'll run this endpoint via wrk, for example, as wrk -c2 -d5s http://localhost:9090/v0.1/posts/userlist/foo123456?p=1&ipp=12 you'll be able to get the error. Maybe concurrency?

Sample code:

var err error
    var IDs []string

    data, _, _, err := g.DB.QueryNeoAll(`
        MATCH (u:User)-[:POSTED]->(p:Post)
        WHERE u.id = {userId}
        RETURN p.id AS id
        ORDER BY p.created DESC
        SKIP {skip}
        LIMIT {limit}
        `,
        map[string]interface{}{
            "limit":  ipp,
            "skip":   page,
            "userId": userID,
        })
    if err != nil {
        return IDs, err
    }

    for _, row := range data {
        IDs = append(IDs, row[0].(string))
    }
    fmt.Printf("IDs: %#v\n", IDs)

    return IDs, nil

Here's how I instantiate the connection to be available to all packages of my application:

var (
    DB bolt.Conn
)

func Connect() error {
    var err error
    driver := bolt.NewDriver()
    DB, err = driver.OpenNeo(config.Get("neo4j.connString"))

    return err
}

Any ideas are appreciated!

Go 1.7.5 Neo4j Community 3.1.1 via official "neo4j" docker image.

Thanks D

[johnnadratowski]

@mpmlj - Sorry for the late response here. This is an interesting one. Do you have a full stack trace, and/or an example script?

[tristanmccarthy]

EDIT: I encountered the same problem with very similar code to the original issue when trying to go from a quick and dirty demo based on the examples to something a bit more usable.

In my case the problem was just blindly following the example code (being new to go). I ended up with concurrent calls to a single bolt.Conn which resulted in the 'open statement' error.

I had a quick read of the docs and switched to using the driver pool. Doing this my code went from:

type NeoDBClient struct {
    conn neoBolt.Conn
}

func (nc *NeoDBClient) ConnectToDB() error {
    var err error
        driver := neoBolt.NewDriver()
    nc.conn, err = driver.OpenNeo("bolt://neo:7687")
    return err
}

func (nc *NeoDBClient) DoThing1() error {
    cypher := "MATCH (n) RETURN n"
    data, _, _, err := nc.conn.QueryNeoAll(cypher, nil)
    return err
}

func (nc *NeoDBClient) DoThing2() error {
    cypher := "MATCH (n) RETURN n"
    data, _, _, err := nc.conn.QueryNeoAll(cypher, nil)
    return err
}

to:

type NeoDBClient struct {
    driverPool neoBolt.DriverPool
}

func (nc *NeoDBClient) ConnectToDB() error {
    var err error
    nc.driverPool, err = neoBolt.NewDriverPool("bolt://neo:7687", 100)
    return err
}

func (nc *NeoDBClient) DoThing1() error {
    conn, err := nc.driverPool.OpenPool()
    cypher := "MATCH (n) RETURN n"
    data, _, _, err := conn.QueryNeoAll(cypher, nil)
    return err
}

func (nc *NeoDBClient) DoThing2() error {
    conn, err := nc.driverPool.OpenPool()
    cypher := "MATCH (n) RETURN n"
    data, _, _, err := conn.QueryNeoAll(cypher, nil)
    return err
}

Not sure this is necessarily right, but it works for now and gets me past the error!

[sashati]

I have the same situation. As I found we are using Thread unsafe Connection method in a mutithread environment. Which is not going to work properly, even by using connection pool. I recommend to synchronize usage of each connection. For example:

var _driver neoBolt.Driver
var _conn neoBolt.Conn

func init() {
    _driver = neoBolt.NewDriver()
    _conn, _ = _driver.OpenNeo(_boltURL)
}

var m sync.Mutex
func getData(query string, params map[string]interface{}) string{

    m.Lock()
    data, _, _, err  := _conn.QueryNeoAll(query, params)
    m.Unlock()
        ...
    return string(resp)
}

[sh0umik]

`` This Problem Simply Means The Connection Is Still Opens and Bolt is not able to Prepare another connection if one still open.

You have to Close the connection After preparing and executing Pipeline or Batch Query

    err = pipeline.Close() // Or Batch Execution Object
    if err != nil {
        panic(err)
    }

That will fix the problem

[collisonchris]

I've run into this a few times and carefully looking at how I was setting up the execution lead me to follow the same path @sh0umik did, closing the statment object before executing anther is the best way to deal with this issue. I ended up writing a generic parameterized execution function:

func executeParameterizedQuery(conn bolt.Conn, queryString string, params map[string]interface{}) (bolt.Result, error) {
    stmt, err := conn.PrepareNeo(queryString)
    if err != nil {
        log.Error(err)
        return nil, err
    }
    result, e := stmt.ExecNeo(params)
    if e != nil {
        log.Error(e)
        return nil, e
    }
    stmt.Close()
    return result, nil
}

[vikashvverma]

I've run into this a few times and carefully looking at how I was setting up the execution lead me to follow the same path @sh0umik did, closing the statment object before executing anther is the best way to deal with this issue. I ended up writing a generic parameterized execution function:

func executeParameterizedQuery(conn bolt.Conn, queryString string, params map[string]interface{}) (bolt.Result, error) {
  stmt, err := conn.PrepareNeo(queryString)
  if err != nil {
      log.Error(err)
      return nil, err
  }
  result, e := stmt.ExecNeo(params)
  if e != nil {
      log.Error(e)
      return nil, e
  }
  stmt.Close()
  return result, nil
}

Are you still able to read the data? I believe if you close the statement then you won't be able to read the result.