More MongoDB bulkWrite() improvements.
Ylian Saint-Hilaire committed
Jan 10, 2021 at 06:19 UTC
b6a257c1e449750da8e3ff87269f04bce3e0a137
1 file changed
+96
-17
db.js
+96
-17
@@ -38,8 +38,20 @@ module.exports.CreateDB = function (parent, func) {
38
obj.dbRecordsDecryptKey = null;
39
obj.changeStream = false;
40
obj.pluginsActive = ((parent.config) && (parent.config.settings) && (parent.config.settings.plugins != null) && (parent.config.settings.plugins != false) && ((typeof parent.config.settings.plugins != 'object') || (parent.config.settings.plugins.enabled != false)));
41
- obj.pendingSet = false;
42
- obj.pendingSets = null;
41
+
42
+ // MongoDB bulk write state
43
+ obj.filePendingSet = false;
44
+ obj.filePendingSets = null;
45
+ obj.filePendingCb = null;
46
+ obj.filePendingCbs = null;
47
+ obj.powerFilePendingSet = false;
48
+ obj.powerFilePendingSets = null;
49
+ obj.powerFilePendingCb = null;
50
+ obj.powerFilePendingCbs = null;
51
+ obj.eventsFilePendingSet = false;
52
+ obj.eventsFilePendingSets = null;
53
+ obj.eventsFilePendingCb = null;
54
+ obj.eventsFilePendingCbs = null;
55
56
obj.SetupDatabase = function (func) {
57
// Check if the database unique identifier is present
@@ -1031,15 +1043,17 @@ module.exports.CreateDB = function (parent, func) {
1043
}
1044
} else if (obj.databaseType == 3) {
1045
// Database actions on the main collection (MongoDB)
1034
- obj.Set = function (data) { // Fast Set operation using bulkWrite(), this is much faster then using replaceOne()
1035
- if (obj.pendingSet == false) {
1046
+ obj.Set = function (data, func) { // Fast Set operation using bulkWrite(), this is much faster then using replaceOne()
1047
+ if (obj.filePendingSet == false) {
1048
// Perform the operation now
1037
- obj.pendingSet = true; obj.pendingSets = null;
1038
- obj.file.bulkWrite([{ replaceOne: { filter: { _id: data._id }, replacement: performTypedRecordEncrypt(common.escapeLinksFieldNameEx(data)), upsert: true } }], bulkWriteCompleted);
1049
+ obj.filePendingSet = true; obj.filePendingSets = null;
1050
+ if (func != null) { obj.filePendingCbs = [func]; }
1051
+ obj.file.bulkWrite([{ replaceOne: { filter: { _id: data._id }, replacement: performTypedRecordEncrypt(common.escapeLinksFieldNameEx(data)), upsert: true } }], fileBulkWriteCompleted);
1052
} else {
1053
// Add this operation to the pending list
1041
- if (obj.pendingSets == null) { obj.pendingSets = {} }
1042
- obj.pendingSets[data._id] = data;
1054
+ if (obj.filePendingSets == null) { obj.filePendingSets = {} }
1055
+ obj.filePendingSets[data._id] = data;
1056
+ if (func != null) { if (obj.filePendingCb == null) { obj.filePendingCb = [ func ]; } else { obj.filePendingCb.push(func); } }
1057
}
1058
};
1059
obj.Get = function (id, func) {
@@ -1113,7 +1127,19 @@ module.exports.CreateDB = function (parent, func) {
1127
1128
// Database actions on the events collection
1129
obj.GetAllEvents = function (func) { obj.eventsfile.find({}).toArray(func); };
1116
- obj.StoreEvent = function (event, func) { obj.eventsfile.insertOne(event, func); };
1130
+ obj.StoreEvent = function (event, func) { // Fast MongoDB event store using bulkWrite()
1131
+ if (obj.eventsFilePendingSet == false) {
1132
+ // Perform the operation now
1133
+ obj.eventsFilePendingSet = true; obj.eventsFilePendingSets = null;
1134
+ if (func != null) { obj.eventsFilePendingCbs = [func]; }
1135
+ obj.eventsfile.bulkWrite([{ insertOne: { document: event } }], eventsFileBulkWriteCompleted);
1136
+ } else {
1137
+ // Add this operation to the pending list
1138
+ if (obj.eventsFilePendingSets == null) { obj.eventsFilePendingSets = [] }
1139
+ obj.eventsFilePendingSets.push(event);
1140
+ if (func != null) { if (obj.eventsFilePendingCb == null) { obj.eventsFilePendingCb = [func]; } else { obj.eventsFilePendingCb.push(func); } }
1141
+ }
1142
+ };
1143
obj.GetEvents = function (ids, domain, func) { obj.eventsfile.find({ domain: domain, ids: { $in: ids } }).project({ type: 0, _id: 0, domain: 0, ids: 0, node: 0 }).sort({ time: -1 }).toArray(func); };
1144
obj.GetEventsWithLimit = function (ids, domain, limit, func) { obj.eventsfile.find({ domain: domain, ids: { $in: ids } }).project({ type: 0, _id: 0, domain: 0, ids: 0, node: 0 }).sort({ time: -1 }).limit(limit).toArray(func); };
1145
obj.GetUserEvents = function (ids, domain, username, func) { obj.eventsfile.find({ domain: domain, $or: [{ ids: { $in: ids } }, { username: username }] }).project({ type: 0, _id: 0, domain: 0, ids: 0, node: 0 }).sort({ time: -1 }).toArray(func); };
@@ -1133,7 +1159,20 @@ module.exports.CreateDB = function (parent, func) {
1159
1160
// Database actions on the power collection
1161
obj.getAllPower = function (func) { obj.powerfile.find({}).toArray(func); };
1136
- obj.storePowerEvent = function (event, multiServer, func) { if (multiServer != null) { event.server = multiServer.serverid; } obj.powerfile.insertOne(event, func); };
1162
+ obj.storePowerEvent = function (event, multiServer, func) { // Fast MongoDB event store using bulkWrite()
1163
+ if (multiServer != null) { event.server = multiServer.serverid; }
1164
+ if (obj.powerFilePendingSet == false) {
1165
+ // Perform the operation now
1166
+ obj.powerFilePendingSet = true; obj.powerFilePendingSets = null;
1167
+ if (func != null) { obj.powerFilePendingCbs = [func]; }
1168
+ obj.powerfile.bulkWrite([{ insertOne: { document: event } }], powerFileBulkWriteCompleted);
1169
+ } else {
1170
+ // Add this operation to the pending list
1171
+ if (obj.powerFilePendingSets == null) { obj.powerFilePendingSets = [] }
1172
+ obj.powerFilePendingSets.push(event);
1173
+ if (func != null) { if (obj.powerFilePendingCb == null) { obj.powerFilePendingCb = [func]; } else { obj.powerFilePendingCb.push(func); } }
1174
+ }
1175
+ };
1176
obj.getPowerTimeline = function (nodeid, func) { obj.powerfile.find({ nodeid: { $in: ['*', nodeid] } }).project({ _id: 0, nodeid: 0, s: 0 }).sort({ time: 1 }).toArray(func); };
1177
obj.removeAllPowerEvents = function () { obj.powerfile.deleteMany({}, { multi: true }); };
1178
obj.removeAllPowerEventsForNode = function (nodeid) { obj.powerfile.deleteMany({ nodeid: nodeid }, { multi: true }); };
@@ -1421,16 +1460,56 @@ module.exports.CreateDB = function (parent, func) {
1460
}
1461
1462
// MongoDB pending bulk write operation, perform fast bulk document replacement.
1424
- function bulkWriteCompleted() {
1425
- if (obj.pendingSets != null) {
1463
+ function fileBulkWriteCompleted() {
1464
+ // Callbacks
1465
+ if (obj.filePendingCbs != null) { for (var i in obj.filePendingCbs) { obj.filePendingCbs[i](); } obj.filePendingCbs = null; }
1466
+ if (obj.filePendingSets != null) {
1467
+ // Perform pending operations
1468
+ var ops = [];
1469
+ obj.filePendingCbs = obj.filePendingCb;
1470
+ obj.filePendingCb = null;
1471
+ for (var i in obj.filePendingSets) { ops.push({ replaceOne: { filter: { _id: i }, replacement: performTypedRecordEncrypt(common.escapeLinksFieldNameEx(obj.filePendingSets[i])), upsert: true } }); }
1472
+ obj.file.bulkWrite(ops, fileBulkWriteCompleted);
1473
+ obj.filePendingSets = null;
1474
+ } else {
1475
+ // All done, no pending operations.
1476
+ obj.filePendingSet = false;
1477
+ }
1478
+ }
1479
+
1480
+ // MongoDB pending bulk write operation, perform fast bulk document replacement.
1481
+ function eventsFileBulkWriteCompleted() {
1482
+ // Callbacks
1483
+ if (obj.eventsFilePendingCbs != null) { for (var i in obj.eventsFilePendingCbs) { obj.eventsFilePendingCbs[i](); } obj.eventsFilePendingCbs = null; }
1484
+ if (obj.eventsFilePendingSets != null) {
1485
+ // Perform pending operations
1486
+ var ops = [];
1487
+ for (var i in obj.eventsFilePendingSets) { ops.push({ document: obj.eventsFilePendingSets[i] }); }
1488
+ obj.eventsFilePendingCbs = obj.eventsFilePendingCb;
1489
+ obj.eventsFilePendingCb = null;
1490
+ obj.eventsFilePendingSets = null;
1491
+ obj.eventsfile.bulkWrite(ops, eventsFileBulkWriteCompleted);
1492
+ } else {
1493
+ // All done, no pending operations.
1494
+ obj.eventsFilePendingSet = false;
1495
+ }
1496
+ }
1497
+
1498
+ // MongoDB pending bulk write operation, perform fast bulk document replacement.
1499
+ function powerFileBulkWriteCompleted() {
1500
+ // Callbacks
1501
+ if (obj.powerFilePendingCbs != null) { for (var i in obj.powerFilePendingCbs) { obj.powerFilePendingCbs[i](); } obj.powerFilePendingCbs = null; }
1502
+ if (obj.powerFilePendingSets != null) {
1503
// Perform pending operations
1427
- var ops = [], c = 0;
1428
- for (var i in obj.pendingSets) { c++; ops.push({ replaceOne: { filter: { _id: i }, replacement: performTypedRecordEncrypt(common.escapeLinksFieldNameEx(obj.pendingSets[i])), upsert: true } }); }
1429
- obj.file.bulkWrite(ops, bulkWriteCompleted);
1430
- obj.pendingSets = null;
1504
+ var ops = [];
1505
+ for (var i in obj.powerFilePendingSets) { ops.push({ document: obj.powerFilePendingSets[i] }); }
1506
+ obj.powerFilePendingCbs = obj.powerFilePendingCb;
1507
+ obj.powerFilePendingCb = null;
1508
+ obj.powerFilePendingSets = null;
1509
+ obj.powerfile.bulkWrite(ops, powerFileBulkWriteCompleted);
1510
} else {
1511
// All done, no pending operations.
1433
- obj.pendingSet = false;
1512
+ obj.powerFilePendingSet = false;
1513
}
1514
}
1515