Nebula Graph: Client Communication Secrets - FBTHRIFT

An overview of the

Nebula Clients provides users with a variety of programming language apis for interacting with Nebula Graph, and repackages the data structures returned by the server for easy use.

Nebula Clients currently supports C++, Java, Python, Golang, and Rust.

Communication framework

Nebula Clients uses fbthrift github.com/facebook/fb… As RPC communication framework between server and client, it realizes cross-language interaction.

Fbthrift provides three functions:

Generated code: FbTHRIFT serializes different languages into data structures
Serialization: Serializes the generated data structure
Communication and interaction: Messages are transmitted between clients and servers, and corresponding server functions are invoked when requests from clients in different languages are received

example

Here is an example of fBThrift in use with Nebula Graph using the Golang client.

VertexStructure definition on the server side:

struct Vertex {
	Value vid;
	std::vector<Tag> tags;

	Vertex() = default;
};
Copy the code

First, insrc/interface/common.thriftDefine some data structures in:

struct Tag {
		1: binary name,
		// List of <prop_name, prop_value>
		2: map<binary, Value> (cpp.template = "std::unordered_map") props,
} (cpp.type = "nebula::Tag")

struct Vertex {
		1: Value     vid,
		2: list<Tag> tags,
} (cpp.type = "nebula::Vertex")
Copy the code

Here we define a Vertex structure with the (CPp. type = “Nebula ::Vertex”) annotation that corresponds to the nebula::Vertex server structure.

Fbthrift automatically generates Golang’s data structure for us:

// Attributes:
// - Vid
// - Tags
type Vertex struct {
	Vid *Value `thrift:"vid,1" db:"vid" json:"vid"`
	Tags []*Tag `thrift:"tags,2" db:"tags" json:"tags"`
}

func NewVertex(a) *Vertex {
	return &Vertex{}
}

...

func (p *Vertex) Read(iprot thrift.Protocol) error { // deserialize. }func (p *Vertex) Write(oprot thrift.Protocol) error { / / the serialization. }Copy the code

inMATCH (v:Person) WHERE id(v) == "ABC" RETURN vThe client requests a vertex from the server.nebula::Vertex), when the server finds the vertexserializationThrough the RPC communication frameworktransportSend to the client, and when the client receives the data, it doesdeserializationGenerate data structures defined in the corresponding client (type Vertex struct).

Client module

In this section, you’ll see the various modules of the client and their main interfaces, using nebula- Go as an example.

The Configs class provides global configuration options.

type PoolConfig struct {
	// Set the timeout period. 0 indicates no timeout, expressed in ms. The default is 0
	TimeOut time.Duration
	// The maximum idle time for each connection. If a connection is not in use within this time, it will be disconnected and deleted. 0 indicates that the connection is permanently idle and will not be closed. The default is 0
	IdleTime time.Duration
	// max_connection_pool_size: Sets the maximum number of connections in the connection pool. The default is 10
	MaxConnPoolSize int
	// Minimum number of free connections. Default: 0
	MinConnPoolSize int
}
Copy the code

Client Session, which provides an interface for users to call directly.

// Manage session-specific information
type Session struct {
	// Used for authentication or message retry during command execution
	sessionID  int64
	// The connection currently held
	connection *connection
	// The connection pool currently in use
	connPool   *ConnectionPool
	// Log tool
	log        Logger
	// Saves the time zone used by the current Session
	timezoneInfo
}
Copy the code

The interface definitions are as follows

	// Execute nGQL. The data type returned is ResultSet. This interface is non-thread-safe.
	func (session *Session) Execute(stmt string) (*ResultSet, error){... }// Retrieve the connection from the connection pool for the current Session
	func (session *Session) reConnect(a) error{... }// Release Session ID and return connection to pool
	func (session *Session) Release(a) {
Copy the code

ConnectionPool, which manages all connections. The main interfaces are as follows

// Create a new connection pool and complete the initialization with the entered service address
func NewConnectionPool(addresses []HostAddress, conf PoolConfig, log Logger) (*ConnectionPool, error){... }// Validate and get the Session instance
func (pool *ConnectionPool) GetSession(username, password string) (*Session, error){... }Copy the code

The Connection, which encapsulates thrift’s network, provides the following interface

// Establish a connection with the specified IP address and port
func (cn *connection) open(hostAddress HostAddress, timeout time.Duration) error{... }// Verify the user name and password
func (cn *connection) authenticate(username, password string) (*graph.AuthResponse, error) {
/ / query execution
func (cn *connection) execute(sessionID int64, stmt string) (*graph.ExecutionResponse, error){... }// Send "YIELD 1" with SessionId 0 to determine if the connection is available
func (cn *connection) ping(a) bool{... }// Release the sessionId to graphd
func (cn *connection) signOut(sessionID int64) error{... }// Disconnect the connection
func (cn *connection) close(a){... }Copy the code

Load balancingLoadBalanceTo use this module in connection pools
- Policy: Polling policy

Module interaction resolution

The connection pool
- Initialization:
  - To use, you need to create and initialize a ConnectionPool. The ConnectionPool is initialized to establish a Connection to the address where the Nebula services are located. If you are deploying multiple Graph services in a clustered deployment, The connection pool uses a polling strategy to balance the load, establishing nearly equal number of connections for each address.
- Management Connection:
  - The connection pool maintains two queues, idleConnectionQueue and idleConnectionQueue. The connection pool periodically detects and closes expired connections. Both queues use read/write locks to ensure correct multithreaded execution while adding or deleting elements.
  - When a Session requests a connection from the connection pool, it checks whether any connection is available in the idle connection queue. If any connection is available, it directly returns the connection to the Session for users to use. If no connections are available and the current total number of connections does not exceed the maximum number specified in the configuration, create a new connection to the Session. If the maximum connection limit has been reached, an error is returned.
- Generally, the connection pool needs to be closed only when the program exits, and all connections in the pool are disconnected when closed.
Client session
- Client SessionAfter the connection pool is generated, the user needs to provide the user password for verification. After the verification succeeds, the user obtains a Session instance and communicates with the server through the connection in the Session. The most commonly used interface isexecute()If an error occurs during execution, the client checks for the type of error, or for network reasonsAutomatic reconnectionAnd try to execute the statement again.
- Note that a Session cannot be used by multiple threads at the same time. The correct way is to use multiple threads to apply for multiple sessions, and each thread uses one Session.
- When a Session is released, its connections are put back into the idle connection queue of the connection pool for later reuse by other sessions.
The connection
- Each connection instance is equivalent and can be held by any Session. The purpose of this design is that these connections can be reused by different sessions to reduce the overhead of repeatedly switching Transport on and off.
- The connection sends the client’s request to the server and returns the result to the Session.
User Example

// Initialize connection pool
pool, err := nebula.NewConnectionPool(hostList, testPoolConfig, log)
iferr ! =nil {
	log.Fatal(fmt.Sprintf("Fail to initialize the connection pool, host: %s, port: %d, %s", address, port, err.Error()))
}
// Close all connections in the pool when program exits
defer pool.Close()

// Create session
session, err := pool.GetSession(username, password)
iferr ! =nil {
	log.Fatal(fmt.Sprintf("Fail to create a new session from connection pool, username: %s, password: %s, %s",
		username, password, err.Error()))
}
// Release session and return connection back to connection pool when program exits
defer session.Release()

// Excute a query
resultSet, err := session.Execute(query)
iferr ! =nil {
	fmt.Print(err.Error())
}
Copy the code

Return data structure

The client encapsulates some of the query results returned by the complex server and adds interfaces to facilitate user use.

Basic type of query results	The encapsulated type
Null
Bool
Int64
Double
String
Time	TimeWrapper
Date
DateTime	DateTimeWrapper
List
Set
Map
Vertex	Node
Edge	Relationship
Path	PathWrraper
DateSet	ResultSet
–	Record(for row operations in a ResultSet)

With nebula::Value, it is wrapped as ValueWrapper on the client side and translated into other structures via interfaces. (i.g. node = ValueWrapper.asNode())

Parsing of data structures

MATCH p= (v:player{name:”Tim Duncan”})-[]->(v2) RETURN p= (v:player{name:”Tim Duncan”})-[]->(v2) RETURN p

+----------------------------------------------------------------------------------------------------------------------- ----------------------------------------------------------------------------------------------------+ | p | +----------------------------------------------------------------------------------------------------------------------- ----------------------------------------------------------------------------------------------------+ | <("Tim Duncan" :bachelor{name: "Tim Duncan", speciality: "psychology"} :player{age: 42, name: "Tim Duncan"})<-[:teammate@0 {end_year: 2016, start_year: 2002}]-("Manu Ginobili" :player{age: 41, name: "Manu Ginobili"})> | +----------------------------------------------------------------------------------------------------------------------- ----------------------------------------------------------------------------------------------------+ Got 1 rows (time spent 11550/12009 us)Copy the code

We can see that the result returned contains a row and is of type a path. To obtain the attributes of the path end point (v2), perform the following operations:

// Excute a query
resultSet, _ := session.Execute("MATCH p= (v:player{name:"\"Tim Duncan"\"})-[]->(v2) RETURN p")

// Get the first row of the result with index 0
record, err := resultSet.GetRowValuesByIndex(0)
iferr ! =nil {
	t.Fatalf(err.Error())
}

// Select the cell from the first row
// valInCol0 is of type ValueWrapper
valInCol0, err := record.GetValueByIndex(0)

// Convert ValueWrapper to a PathWrapper object
pathWrap, err = valInCol0.AsPath()

// Get the endpoint directly through the GetEndNode() interface of PathWrapper
node, err = pathWrap.GetEndNode()

// Get all Properties from Node Properties()
// The type of the props is map[string]*ValueWrapper
props, err = node.Properties()
Copy the code

Client address

GitHub address of each language client:

Github.com/vesoft-inc/…
Github.com/vesoft-inc/…
Github.com/vesoft-inc/…
Github.com/vesoft-inc/…
Github.com/vesoft-inc/…

mo4tech.com (Moment For Technology) is a global community with thousands techies from across the global hang out!Passionate technologists, be it gadget freaks, tech enthusiasts, coders, technopreneurs, or CIOs, you would find them all here.

Nebula Graph: Client Communication Secrets – FBTHRIFT

An overview of the

Communication framework

example

Client module

Module interaction resolution

Return data structure

Parsing of data structures

Client address

Recommended reading

Nebula Graph: Client Communication Secrets – FBTHRIFT

An overview of the

Communication framework

example

Client module

Module interaction resolution

Return data structure

Parsing of data structures

Client address

Recommended reading

Related Posts

Kubernetes(K8s)

Interview series 3: Redis

Feel the power of Python asynchronous crawlers (gEvent, async, await)