请教skynet的mongo是否支持bulkWrite接口？我这边做了测试未找到对应方法。

[253980289]

请教skynet的mongo是否支持bulkWrite接口？我这边做了测试未找到对应方法。
我的模仿代码：
local data = {
insertOne = { document = { _id = bson.objectid(), char = "Dithras", class = "barbarian", lvl = 4 } }
}
local r = self.database:runCommand("bulkwrite", self.name, "bulks", data)
return werror(r)

返回结果如下：
{
["code"] = 59,
["codeName"] = "CommandNotFound",
["ok"] = 0.0,
["errmsg"] = "no such command: 'bulkwrite'",
}

[cloudwu]

我不熟悉 mongodb ，但这个问题你应该去查 mongo 的文档。

[253980289]

明白了，谢谢。

[cloudwu]

我查了一下，实现 bulkWrite 需要先创建一个 bulk 对象，然后执行：https://github.com/mongodb/specifications/blob/master/source/driver-bulk-update.rst

可以参考官方 js 的实现：https://github.com/mongodb/mongo/blob/master/src/mongo/shell/bulk_api.js#L914-L927

如果你按协议实现了，欢迎提交 pr 。

[253980289]

我对mongo其实也仅仅只有一点点应用层的使用经验，今天对mongo关于bulkWrite接口做了一些简单的调研：
我先简单研究了一下你发的js版本的实现，它这个我的理解是shell层的客户端包装实现版本，我另外研究了mongoc的源码相关实现，好像也是同样的客户端封装实现，仍然是一条条for循环请求再合并结果的。

   for (i = 0; i < bulk->commands.len; i++) {
      if (bulk->server_id) {
         server_stream =
            mongoc_cluster_stream_for_server (cluster,
                                              bulk->server_id,
                                              true /* reconnect_ok */,
                                              bulk->session,
                                              reply,
                                              error);
      } else {
         server_stream = mongoc_cluster_stream_for_writes (
            cluster, bulk->session, reply, error);
      }

      if (!server_stream) {
         /* stream_for_server and stream_for_writes initialize reply on error */
         RETURN (false);
      }

      command =
         &_mongoc_array_index (&bulk->commands, mongoc_write_command_t, i);

      _mongoc_write_command_execute (command,
                                     bulk->client,
                                     server_stream,
                                     bulk->database,
                                     bulk->collection,
                                     bulk->write_concern,
                                     offset,
                                     bulk->session,
                                     &bulk->result);

      bulk->server_id = server_stream->sd->id;
      /* If a retryable error occurred and a new primary was selected, use it in
       * subsequent commands. */
      if (bulk->result.retry_server_id) {
         bulk->server_id = bulk->result.retry_server_id;
      }

      if (bulk->result.failed &&
          (bulk->flags.ordered || bulk->result.must_stop)) {
         mongoc_server_stream_cleanup (server_stream);
         GOTO (cleanup);
      }

      offset += command->n_documents;
      mongoc_server_stream_cleanup (server_stream);
   }

另外，我也简单看了skynet关于mongo的驱动实现代码lua-mongo.c，有个疑问，为啥不是直接以官方mongoc驱动为基础来封，而是另个实现了一套？

[cloudwu]

mongo driver 的 wire protocol 非常简单，只有这部分才需要用 C 实现。其余部分并不需要从 C 接口导成 lua 接口，也很难一一导入。

另外，mongoc 的设计非常糟糕，它甚至把 socket 通讯层和通讯协议的实现混杂在一起实现的，无法分离。而 lua-mongo.c 只做了 wire protocol 的数据打包工作，没有任何多余的功能。

[cloudwu]

前面我看错了。 bulk update 是已经废弃的 api ，bulk write 属于 crud api 的一部分，应该参考: https://github.com/mongodb/specifications/blob/master/source/crud/crud.rst#insert-update-replace-delete-and-bulk-writes

[cloudwu]

bulk write 目前应该是用 OP_MSG 这条协议实现的 ： https://github.com/mongodb/mongo-c-driver/blob/master/src/libmongoc/src/mongoc/mongoc-write-command.c#L1055

Write protocol 见 https://www.mongodb.com/docs/manual/reference/mongodb-wire-protocol/ 文档说，从 MongoDB 5.0 开始，只使用 OP_MSG 一条协议，取代之前所有。

OP_MSG 目前在 lua-mongo.c 中尚未封装。但封装很简单，就是消息头 + 消息字符串内容。
而此处的消息协议格式，应该在这里有参考：

https://github.com/mongodb/mongo-c-driver/blob/master/src/libmongoc/src/mongoc/mongoc-write-command.c#L518-L527

摘抄如下：

  /*
    * OP_MSG header == 16 byte
    * + 4 bytes flagBits
    * + 1 byte payload type = 1
    * + 1 byte payload type = 2
    * + 4 byte size of payload
    * == 26 bytes opcode overhead
    * + X Full command document {insert: "test", writeConcern: {...}}
    * + Y command identifier ("documents", "deletes", "updates") ( + \0)
    */

即 26 字节的消息头加一个 bson 文档和若干指令字符串。封装 C 接口应该按此协议传入参数，打包成最终的数据包。

[253980289]

我这边咨询过关于bulkWrite接口是基于server端实现还是client实现的问题，有答复说是基于服务端实现的，但他们未能提供足够说服力的证据证明是基于服务端实现的，而我这边做的调查结论都是基于客户端封装的，仍然是将请求解析成一条条原有insert,update,delete操作，然后用for循环一个个执行并拼合结果，而官方也仅仅是在shell命令层实现了js的相应封装。在mongo的github源码的cpp和h文件中也未找到bulkWrite相应函数实现。如果我调研得没错的话，我们是否也需要或值得封装一个这样的客户接口。

[cloudwu]

我了研究一下 https://github.com/mongodb/specifications/blob/master/source/crud/crud.rst CRUD API ，我是这样理解的：

Current drivers have chosen slightly different names and semantics for the same operations and options. In addition, not all drivers offer all the same operations and methods. As such, it is difficult to transition from driver to driver making the jobs of polyglot developers, documentation authors, and support engineers more difficult.

不同的语言的不同 driver 做相同的事情，可能 api 叫不同的名字。为了解决这个问题，mongo 的新版本打算引入统一的 CRUD API 做所有的事情，只是提供的参数不同。所以，以后都推荐使用它。

Q : Didn't we just build a bulk API?
Yes, most drivers did just build out a bulk API (fluent-bulk-api). While unfortunate, we felt it better to have the bulk api be consistent with the rest of the methods in the CRUD family of operations. However, the fluent-bulk-api is still able to be used as this change is non-backwards breaking. Any driver which implemented the fluent bulk API should deprecate it and drivers that have not built it should not do so.

而 CRUD Api 顺便就提供了 bulk 的功能，把多条操作打包在一起发送，并批量获取结果。有了它，driver 就不应该再自己实现 bulk api 。

关于 bulk 我是这样想的。之所以许多 driver 都提供了 bulk 打包多条指令。是因为，大多数 driver 是从 socket 层面做起，并按一问一答 Req/Resp 模式工作的。如果没拿到 server 的 resp 就不会发起下一次 Req 。这样就限制了吞吐量。为了解决这个问题，就再次设计了 bulk api ，批量提起 req 。

但是 skynet 不是这样工作的。skynet 的 IO 并不会阻塞。一条条指令发送和打包发送是没有性能区别的。你用同一个 client 对象，在不同的 coroutine 中发送，就可以做到非阻塞请求。配合 skynet.select 还能简化处理 #1602

[253980289]

也就是说mongo是在未来的新版本中才会在服务端引入bulkWrite接口了，我之前想使用bulkWrite接口，主要是从两个方面考虑，还未涉及到性能层面上：
1、事务层面，希望批量操作能达到原子操作的效果；
2、调用接口的便捷性上，不用反复调用不同接口，而只需调用一个接口，提供各种参数即可。
如果mongo确实是要在未来才提供服务端bulkWrite接口，那我们是否也是在mongo正式提供新接口后再考虑增加bulkWrite接口呢?
目前看，如果只是在客户端做封装，对skynet框架来说，其实意义就不大了。

[ykxpb]

bulkwrite 应该做不到 1，看代码 bulk write 只是一次性打包了多个请求。insert many、update many 类似。

对事务的支持倒是可以搞一下。

https://github.com/mongodb/specifications/blob/b2d215925f4c017d57652c17b136243a47402d5f/source/transactions/transactions.rst

[253980289]

目前mongo提供的接口其实已经提供了混合型(通过upsert, multi参数)批量操作支持，目前缺少的仅是delete操作的混合请求。

[253980289]

经研究和测试发现，目前mongo服务端已支持单请求批量插入和更新同时进行，相对于mogo官方shell客户端封装的bulkWrite来说，少了delete操作，目前skynet的mongo.lua接口中只需增加以下方法即可：

function mongo_collection:safe_update_list(update_list)
	for i,v in ipairs(update_list) do
		update_list[i] = bson_encode(v)
	end
	local r = self.database:runCommand("update", self.name, "updates", update_list)
	return werror(r)
end

测试代码：

        tb_user_prop.safe_update_list(tb_name, {
            {
                q = {_id = 429501024667301},
                u = {["$set"] = {test2 = "123"}},
                upsert = true,
                multi = false,
            },
            {
                q = {_id = 429501024667309},
                u = {test2 = "123"},
                upsert = true,
                multi = false,
            },
        })

调用日志如下：

[sql][2022-07-11 09:54:58.54][0x7fcfb206a608][0981][gdbproxyserver-gcommon-f(981)][ghome]	safe_update_list	tb_user_prop	{
  [1]  =  {
    ["q"]  =  {
      ["_id"]  =  429501024667301,
    },
    ["u"]  =  {
      ["$set"]  =  {
        ["test2"]  =  "123",
      },
    },
    ["multi"]  =  false,
    ["upsert"]  =  true,
  },
  [2]  =  {
    ["q"]  =  {
      ["_id"]  =  429501024667309,
    },
    ["u"]  =  {
      ["test2"]  =  "123",
    },
    ["multi"]  =  false,
    ["upsert"]  =  true,
  },
}[mqlen:0]-log_mqlen:0
[sql][2022-07-11 09:54:58.54][0x7fcfb206a608][0043][gdbproxyserver-gdbcom-double_call(43)][ghome]	true	{
  ["n"]  =  2,
  ["ok"]  =  1.0,
  ["nModified"]  =  2,
}[mqlen:0]-log_mqlen:1

[253980289]

按mongo官方shell脚本的接口形式实现了一个简单的封装版本，对性能有要求的可以按云大说的select方式进一步改进：

function mongo_collection:safe_bulkwrite(bulkwrite_list, ordered)
	local ret = {}
	for i,v in ipairs(bulkwrite_list) do
		for k,v2 in pairs(v) do
			if "insertOne" == k then
				table.insert(ret, self.database:runCommand("insert", self.name, "documents", {bson_encode(v2.document)}))
			elseif "updateOne" == k then
				table.insert(ret, self.database:runCommand("update", self.name, "updates", {bson_encode({
					q = v2.filter,
					u = v2.update,
					upsert = v2.upsert,
					-- multi = nil,
				})}))
			elseif "updateMany" == k then
				table.insert(ret, self.database:runCommand("update", self.name, "updates", {bson_encode({
					q = v2.filter,
					u = v2.update,
					upsert = v2.upsert,
					multi = true,
				})}))
			elseif "replaceOne" == k then
				table.insert(ret, self.database:runCommand("update", self.name, "updates", {bson_encode({
					q = v2.filter,
					u = v2.replacement,
					upsert = v2.upsert,
					-- multi = nil,
				})}))
			elseif "deleteOne" == k then
				table.insert(ret, self.database:runCommand("delete", self.name, "deletes", {bson_encode({
					q = v2.filter,
					limit = 1,
				})}))
			elseif "deleteMany" == k then
				table.insert(ret, self.database:runCommand("delete", self.name, "deletes", {bson_encode({
					q = v2.filter,
					limit = 0,
				})}))
			else
				assert(false, "unknow request " .. k)
			end
		end
	end
	return ret
end

为啥我用不好这个代码编辑功能，所有内容会变成一行，而不用代码编辑功能，代码里的缩进格式又会被去掉。

@zhangshiqian1214
改好了，感谢，封闭太久了^L^

[zhangshiqian1214]

了解一下markdown语法

[253980289]

附测试代码：

        gdata.dlog(gdata.bulkWrite("tb_test", {
            { insertOne = { document = { _id = 4, char = "Dithras", class = "barbarian", lvl = 4 } } },
            { insertOne = { document = { _id = 5, char = "Taeln", class = "fighter", lvl = 3 } } },
            { updateOne  = {
                    filter  = { char  = "Eldon" },
                    update  = { ["$set"]  = { status  = "Critical Injury" } }
                } 
            },
            { replaceOne  = {
                    filter  = { char  = "Meldane" },
                    replacement  = { char  = "Tanys", class  = "oracle", lvl = 4 }
                } 
            },
            { updateMany = {
                    filter = { rating = { ["$lt"] = 2} },
                    update = { ["$inc"] = { encounter = -0.25 } }
                },
            },
            { deleteMany = { filter = { encounter = { ["$lt"] = 0 } } } },
            { deleteOne  = { filter  = { char  = "Brisbane"} } },
        }))

[cloudwu]

可以看看 #1649