-
|
Using the Thread Bucket Selection Strategy and choosing adequate values for the number of buckets and number of async worker threads, it is possible to eliminate contention on insertion. |
Beta Was this translation helpful? Give feedback.
Replies: 8 comments 20 replies
-
|
Obviously, the "brute-force" solution is to synchronize the method that does the update, but I have tried this and it slows down the the update by about 10%. |
Beta Was this translation helpful? Give feedback.
-
|
This is an interesting topic. I have a branch with pessimistic locking to try in case. In your use case how many records are in a transaction? If you have millions of record the chances of updating the same page (max 2K records per page) should be low. How many tx/sec are you able to do right now? Are.you using ArcadeDB embedded or client/server? In the latter case, are you using http? |
Beta Was this translation helpful? Give feedback.
-
|
I use ArcadeDB embedded, in this particular test scenario the db contains very few records (< 1000, so it's a worst-case scenario for the page contention). and I have a multi-threaded client (4 threads in this particular configuration) that sends http requests to an Undertow handler that executes the sql update statement, it is guaranteed that each client thread sends a request to update a different record. |
Beta Was this translation helpful? Give feedback.
-
|
In the class DatabaseAsync, on update the slot is correctly selected by using its RID. So an update without an explicit transaction should always go in parallel without conflict. // UPDATE
final int slot = getSlot(record.getIdentity().getBucketId());
scheduleTask(slot, new DatabaseAsyncUpdateRecord(record, updateRecordCallback, errorCallback), true, backPressurePercentage);The issue could be forcing a tx commit after all update operations. Or I should check the transaction async task if it's smart enough to use the thread slot if all the best conditions are met, namely all the operations are on the same thread-slot (by RID). In your case with one update per TX is possible, it's just an edge case I never considered before. Let me check. |
Beta Was this translation helpful? Give feedback.
-
|
I just pushed a small enhancement with a new API that takes the void transaction(Database.TransactionScope txBlock, int retries, OkCallback ok, ErrorCallback error, int slot);In your case you could do this: final int slot = ((DatabaseAsyncExecutorImpl) database.async()).getSlot( recordToUpdate.getIdentity().getBucketId() );
database.async().transaction( ()-> {
recordToUpdate.modify().set( "X", "Y" ).save();
}, 0, null, null, slot );In practice by using the |
Beta Was this translation helpful? Give feedback.
-
|
How do I obtain the This is what I've come up with, but it looks ugly, with all those lambdas: |
Beta Was this translation helpful? Give feedback.
-
|
Since you're using the Java API, why don't update the record via API instead of SQL? If you use the MutableDocument recordToUpdate = database.lookupByKey("Order", "id", orderId).next().getRecord().asDocument(true).modify();
// IF THE RECORD IS A VERTEX, USE THIS INSTEAD
MutableVertex recordToUpdate = database.lookupByKey("Order", "id", orderId).next().getRecord().asVertex(true).modify();
if( status != null )
recordToUpdate.set( "status", status );
if( pstart != null )
recordToUpdate.set( "pstart", pstart );
if( pstop != null )
recordToUpdate.set( "pstop", pstop );
if( node != null )
recordToUpdate.set( "node", node );
database.async().updateRecord( recordToUpdate, null, (exception)-> {
// MANAGE HERE ANY ERRORS
} ); |
Beta Was this translation helpful? Give feedback.
-
|
Does the Java API use the indexes? |
Beta Was this translation helpful? Give feedback.
-
|
It does use the index and also it selects the right slot for you based on record's id. Check the code I wrote, it should as simple as that. |
Beta Was this translation helpful? Give feedback.
-
|
I have tried that code that I have posted, and I get a lot of Concurrent Modification Exceptions: These are a lot more than I got when I didn't use the async API and I didn't use the automatic transaction with retries, maybe because I have inserted the records using the sync API, so they all ended up in the same bucket. |
Beta Was this translation helpful? Give feedback.
-
|
I thought that maybe the new API works well only if the records have been inserted asynchronously, so I re-wrote my insert method, but now I don't get any records inserted at all. I attach my full method, which produces the following results, when passed a list of 107 objects to insert:
I have also tried to remove the transaction, but I get the same result. |
Beta Was this translation helpful? Give feedback.
-
|
I see you call an async update from an async transaction. The best in terms of performance is to execute the lookup in the main thread and then schedule an insert or an update asynchronously. below you can find a full test case that inserts 100K of orders, then sends another batch of orders, where half are updates and half are new orders. In the end, checks that everything is all right. It looks pretty fast to me, but please let me know how long Hsqldb takes to do the same. /*
* Copyright 2023 Arcade Data Ltd
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package com.arcadedb.server;
import com.arcadedb.ContextConfiguration;
import com.arcadedb.GlobalConfiguration;
import com.arcadedb.database.Database;
import com.arcadedb.database.DatabaseFactory;
import com.arcadedb.database.MutableDocument;
import com.arcadedb.database.Record;
import com.arcadedb.index.IndexCursor;
import com.arcadedb.integration.misc.IntegrationUtils;
import com.arcadedb.schema.DocumentType;
import com.arcadedb.schema.Schema;
import com.arcadedb.schema.Type;
import com.arcadedb.serializer.json.JSONObject;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import static com.arcadedb.server.BaseGraphServerTest.DEFAULT_PASSWORD_FOR_TESTS;
/**
* From Issue https://github.com/ArcadeData/arcadedb/discussion/1129
*/
public class BatchInsertUpdateTest {
static final String DATABASE_NAME = "BatchInsertUpdateTest";
private class CandidateOrder {
private final String[] values;
private CandidateOrder(final String[] values) {
this.values = values;
}
public String getProcessor() {
return values[0];
}
public String getTriggerRid() {
return values[1];
}
public String getStartTime() {
return values[2];
}
public String getStopTime() {
return values[3];
}
public String getStatus() {
return values[4];
}
public void setStatus(String newStatus) {
values[4] = newStatus;
}
}
@Test
public void testBatchAsyncInsertUpdate() {
final ContextConfiguration serverConfiguration = new ContextConfiguration();
final String rootPath = IntegrationUtils.setRootPath(serverConfiguration);
try (DatabaseFactory databaseFactory = new DatabaseFactory(rootPath + "/databases/" + DATABASE_NAME)) {
if (databaseFactory.exists())
databaseFactory.open().drop();
try (Database database = databaseFactory.create()) {
DocumentType type = database.getSchema().getOrCreateDocumentType("Order");
type.createProperty("processor", Type.STRING);
type.createProperty("vstart", Type.STRING);
type.createProperty("vstop", Type.STRING);
type.createTypeIndex(Schema.INDEX_TYPE.LSM_TREE, true, "processor", "vstart", "vstop");
}
ArcadeDBServer arcadeDBServer = new ArcadeDBServer(serverConfiguration);
arcadeDBServer.start();
Database database = arcadeDBServer.getDatabase(DATABASE_NAME);
try {
int TOTAL = 100_000;
final List<CandidateOrder> ordersToProcess = new ArrayList<>(TOTAL);
for (int i = 0; i < TOTAL; i++)
ordersToProcess.add(new CandidateOrder(new String[] { "" + i, "" + i, "" + i, "" + i, "created" }));
insertOrdersAsync(database, ordersToProcess);
for (int i = 0; i < TOTAL / 2; i++)
ordersToProcess.get(i).setStatus("updated");
for (int i = TOTAL / 2; i < TOTAL; i++) {
int k = TOTAL + i;
ordersToProcess.set(i, new CandidateOrder(new String[] { "" + k, "" + k, "" + k, "" + k, "created" }));
}
insertOrdersAsync(database, ordersToProcess);
int created = 0;
int updated = 0;
for (Iterator<Record> it = database.iterateType("Order", false); it.hasNext(); ) {
final String status = it.next().asDocument().getString("status");
if (status.equals("created"))
++created;
else if (status.equals("updated"))
++updated;
}
Assertions.assertEquals(TOTAL, created);
Assertions.assertEquals(TOTAL / 2, updated);
} finally {
arcadeDBServer.stop();
databaseFactory.open().drop();
}
}
}
private JSONObject insertOrdersAsync(final Database database, List<CandidateOrder> orders) {
final long begin = System.currentTimeMillis();
JSONObject result = new JSONObject();
CountDownLatch countDownLatch = new CountDownLatch(orders.size());
AtomicLong counter = new AtomicLong();
int[] firstOrderId = new int[1];
database.async().onError(exception -> {
exception.printStackTrace();
});
final AtomicLong autoIncrementOrderId = new AtomicLong();
for (CandidateOrder order : orders) {
IndexCursor indexCursor = database.lookupByKey("Order", new String[] { "processor", "vstart", "vstop" },
new Object[] { order.getProcessor(), order.getStartTime(), order.getStopTime() });
MutableDocument record;
if (indexCursor.hasNext()) {
record = indexCursor.next().getRecord().asDocument(true).modify();
record.set("id", autoIncrementOrderId.incrementAndGet());
record.set("processor", order.getProcessor());
record.set("trigger", order.getTriggerRid());
record.set("vstart", order.getStartTime());
record.set("vstop", order.getStopTime());
record.set("status", order.getStatus());
database.async().updateRecord(record, newRecord -> {
counter.incrementAndGet();
countDownLatch.countDown();
}, exception -> {
exception.printStackTrace();
});
} else {
record = database.newDocument("Order");
record.set("id", autoIncrementOrderId.incrementAndGet());
record.set("processor", order.getProcessor());
record.set("trigger", order.getTriggerRid());
record.set("vstart", order.getStartTime());
record.set("vstop", order.getStopTime());
record.set("status", order.getStatus());
database.async().createRecord(record, newRecord -> {
counter.incrementAndGet();
countDownLatch.countDown();
}, exception -> {
exception.printStackTrace();
autoIncrementOrderId.decrementAndGet();
});
}
}
database.async().waitCompletion(10_000);
result.put("totalRows", counter.get());
System.out.println("insert orders result = " + result + " in " + (System.currentTimeMillis() - begin) + "ms");
return result;
}
@BeforeAll
public static void beginTests() {
final ContextConfiguration serverConfiguration = new ContextConfiguration();
final String rootPath = IntegrationUtils.setRootPath(serverConfiguration);
GlobalConfiguration.SERVER_ROOT_PASSWORD.setValue(DEFAULT_PASSWORD_FOR_TESTS);
//GlobalConfiguration.TYPE_DEFAULT_BUCKETS.setValue(1);
try (DatabaseFactory databaseFactory = new DatabaseFactory(rootPath + "/databases/" + DATABASE_NAME)) {
if (databaseFactory.exists())
databaseFactory.open().drop();
}
}
@AfterAll
public static void endTests() {
TestServerHelper.checkActiveDatabases();
GlobalConfiguration.resetAll();
}
} |
Beta Was this translation helpful? Give feedback.
-
|
Your test works because you use If modify your test case to wait for the count down latch, and I put TOTAL = 100, I get: |
Beta Was this translation helpful? Give feedback.
-
|
Could you please send the modified test so I can check it out? |
Beta Was this translation helpful? Give feedback.
-
|
Attached. |
Beta Was this translation helpful? Give feedback.
-
|
Yes, without the call to You can also change that to commit every time, just use this at the beginning For your use case, if I understand correctly, since you have these batches of updates, the best way is to do as with my test: one thread checks if the record exists, and then it schedules create or update in async tasks. When everything is sent, you just wait for the finish. At that point everything is committed in the database. |
Beta Was this translation helpful? Give feedback.
-
|
How fast can you do the same with Hsqldb? I doubt you can get close to this speed, especially because it's hard to squeeze all the cores. if you check the CPU usage during ArcadeDB and HSQLDB you should see a noticeable difference in that too. |
Beta Was this translation helpful? Give feedback.
-
|
I still cannot get it to work in my project. |
Beta Was this translation helpful? Give feedback.
-
|
Ok, eager to see a test case that reproduces the issue. |
Beta Was this translation helpful? Give feedback.
-
|
Ok, I made some progress. |
Beta Was this translation helpful? Give feedback.
-
|
and finally I have a test case that allows to reproduce the issue using the Console. Interestingly, if I execute the test on my laptop (8 cores), setting the number of buckets and the number of async worker threads to 6, the select from the Console returns 600 records out of 3000, but if I execute the test on the Customer's test server (24 cores), setting the number of buckets and the number of async worker threads to 8, the select from the Console returns 100 records out of 3000. I don't know if it means anything. |
Beta Was this translation helpful? Give feedback.
-
|
Pulled latest snapshot, cleaned all IDE caches and re-built indexes, re-compiled everything and now both your test case and my test case pass. |
Beta Was this translation helpful? Give feedback.
-
|
Good it works! |
Beta Was this translation helpful? Give feedback.
-
|
I have replicated the same test case with Hsqldb, set up for maximum performance:
and inserting 100K records takes 1.8 seconds. The equivalent ArcadeDB test case shows a strong dependency on the use of the SQL vs Java API, and a weak dependency on the number of cores and the queue implementation. So the take-away message is: in this particular scenario the Java API is much faster than the SQL API. Of course Hsqldb has other drawbacks with respect to ArcadeDB, in particular it does not scale well (all the tables and indexes are stored in a single file). |
Beta Was this translation helpful? Give feedback.
-
|
Interesting results. I tried to play a little bit with your code and I notice a big improvement if you use the partitioned strategy: dtProducts.setBucketSelectionStrategy(new PartitionedBucketSelectionStrategy(List.of("name")));I was able to insert 100K products (considering also 2 indexes) in 1.4 secs on my Mac (parallel level = 6). Below the test case in case you need to reproduce it: package com.arcadedb.console;
import com.arcadedb.ContextConfiguration;
import com.arcadedb.GlobalConfiguration;
import com.arcadedb.database.Database;
import com.arcadedb.database.DatabaseFactory;
import com.arcadedb.database.MutableDocument;
import com.arcadedb.database.bucketselectionstrategy.PartitionedBucketSelectionStrategy;
import com.arcadedb.index.IndexCursor;
import com.arcadedb.query.sql.executor.Result;
import com.arcadedb.query.sql.executor.ResultSet;
import com.arcadedb.schema.DocumentType;
import com.arcadedb.schema.Schema;
import com.arcadedb.schema.Type;
import com.arcadedb.serializer.json.JSONArray;
import com.arcadedb.serializer.json.JSONObject;
import com.arcadedb.server.ArcadeDBServer;
import com.arcadedb.server.TestServerHelper;
import com.arcadedb.server.security.ServerSecurity;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.io.*;
import java.time.*;
import java.util.*;
import java.util.concurrent.atomic.*;
import static com.arcadedb.server.StaticBaseServerTest.DEFAULT_PASSWORD_FOR_TESTS;
/**
* From Discussion https://github.com/ArcadeData/arcadedb/discussions/1129#discussioncomment-6226545
*/
public class ConsoleAsyncInsertTest {
static final String DATABASE_NAME = "ConsoleAsyncInsertTest";
static final int PARALLEL_LEVEL = 6;
private static class Product {
private final String fileName;
private final String fileType;
private final LocalDateTime startValidity;
private final LocalDateTime stopValidity;
private final String version;
public Product(String fileName, String fileType, LocalDateTime startValidity, LocalDateTime stopValidity, String version) {
this.fileName = fileName;
this.fileType = fileType;
this.startValidity = startValidity;
this.stopValidity = stopValidity;
this.version = version;
}
public String getFileName() {
return fileName;
}
public String getFileType() {
return fileType;
}
public LocalDateTime getStartValidity() {
return startValidity;
}
public LocalDateTime getStopValidity() {
return stopValidity;
}
public String getVersion() {
return version;
}
}
@Test
public void testBulkAsyncInsertProducts() {
GlobalConfiguration.SERVER_ROOT_PATH.setValue(".");
GlobalConfiguration.SERVER_DATABASE_DIRECTORY.setValue("databases");
try (DatabaseFactory databaseFactory = new DatabaseFactory("databases/" + DATABASE_NAME)) {
if (databaseFactory.exists()) {
databaseFactory.open().drop();
}
try (Database db = databaseFactory.create()) {
db.transaction(() -> {
DocumentType dtProducts = db.getSchema().buildDocumentType().withName("Product").withTotalBuckets(PARALLEL_LEVEL).create();
dtProducts.createProperty("name", Type.STRING);
dtProducts.createProperty("type", Type.STRING);
dtProducts.createProperty("start", Type.DATETIME_MICROS);
dtProducts.createProperty("stop", Type.DATETIME_MICROS);
dtProducts.createProperty("v", Type.STRING);
dtProducts.createTypeIndex(Schema.INDEX_TYPE.LSM_TREE, true, "name");
dtProducts.createTypeIndex(Schema.INDEX_TYPE.LSM_TREE, false, "type", "start", "stop");
//dtProducts.setBucketSelectionStrategy(new ThreadBucketSelectionStrategy());
dtProducts.setBucketSelectionStrategy(new PartitionedBucketSelectionStrategy(List.of("name")));
Assertions.assertEquals(PARALLEL_LEVEL, dtProducts.getBuckets(false).size());
});
}
}
String dateTimePattern = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS";
GlobalConfiguration.SERVER_METRICS.setValue(false);
GlobalConfiguration.HA_ENABLED.setValue(false);
GlobalConfiguration.DATE_TIME_IMPLEMENTATION.setValue(java.time.LocalDateTime.class);
GlobalConfiguration.DATE_TIME_FORMAT.setValue(dateTimePattern);
GlobalConfiguration.ASYNC_WORKER_THREADS.setValue(PARALLEL_LEVEL);
ContextConfiguration configuration = new ContextConfiguration();
configuration.setValue(GlobalConfiguration.SERVER_ROOT_PASSWORD, DEFAULT_PASSWORD_FOR_TESTS);
AtomicLong txErrorCounter = new AtomicLong();
ArcadeDBServer arcadeDBServer = new ArcadeDBServer(configuration);
arcadeDBServer.start();
ServerSecurity serverSecurity = arcadeDBServer.getSecurity();
final String userName = "root";
final String password = com.arcadedb.server.BaseGraphServerTest.DEFAULT_PASSWORD_FOR_TESTS;
if (serverSecurity.getUser(userName) == null) {
serverSecurity.createUser(new JSONObject().put("name", userName).put("password", serverSecurity.encodePassword(password))
.put("databases", new JSONObject().put(DATABASE_NAME, new JSONArray(new String[] { "admin" }))));
}
Database database = arcadeDBServer.getDatabase(DATABASE_NAME);
database.async().setParallelLevel(PARALLEL_LEVEL);
database.async().onError(exception -> {
System.out.println("database.async() error: " + exception.getMessage());
exception.printStackTrace();
txErrorCounter.incrementAndGet();
});
final AtomicLong okCount = new AtomicLong();
final AtomicLong errCount = new AtomicLong();
String name;
final long N = 100_000;
Product product;
LocalDateTime start, stop, ref;
ref = LocalDateTime.now().minusMonths(1);
final long begin = System.currentTimeMillis();
for (int i = 0; i < N; i++) {
name = UUID.randomUUID().toString();
start = ref.plusMinutes(i);
stop = ref.plusMinutes(i + 1);
product = new Product(name, "SIR1LRM_0_", start, stop, "0001");
inventoryProductAsync(database, product, okCount, errCount);
}
database.async().waitCompletion();
System.out.println("Total async insertion of " + N + " elements in " + (System.currentTimeMillis() - begin));
try {
Assertions.assertTrue(database.async().waitCompletion(30_000));
Assertions.assertEquals(okCount.get(), N);
Assertions.assertEquals(errCount.get(), 0);
Assertions.assertEquals(txErrorCounter.get(), 0);
try (ResultSet resultSet = database.query("sql", "SELECT count(*) as total FROM Product")) {
Result result = resultSet.next();
Assertions.assertEquals((Long) result.getProperty("total"), N);
Console console = new Console();
String URL = "remote:localhost/" + DATABASE_NAME + " " + userName + " " + password;
Assertions.assertTrue(console.parse("connect " + URL));
final StringBuilder buffer = new StringBuilder();
console.setOutput(output -> buffer.append(output));
Assertions.assertTrue(console.parse("select count(*) from Product"));
String[] lines = buffer.toString().split("\\r?\\n|\\r");
int count = Integer.parseInt(lines[4].split("\\|")[2].trim());
Assertions.assertEquals(N, count);
} catch (IOException e) {
System.out.println(e.getMessage());
}
} finally {
arcadeDBServer.stop();
}
}
private void inventoryProductAsync(Database database, Product p, AtomicLong okCount, AtomicLong errCount) {
String fileName = p.getFileName();
try {
final IndexCursor cursor = database.lookupByKey("Product", "name", fileName);
if (cursor.hasNext()) {
final MutableDocument record = cursor.next().asDocument(true).modify();
database.async().updateRecord(record, updatedRecord -> okCount.incrementAndGet(), exception -> {
errCount.incrementAndGet();
System.out.println("database.async.command() error: " + exception.getMessage());
});
} else {
final MutableDocument record = database.newDocument("Product")
.set("name", fileName, "type", p.getFileType(), "start", p.getStartValidity(), "stop", p.getStopValidity(), "v", p.getVersion());
database.async().createRecord(record, newRecord -> okCount.incrementAndGet(), exception -> {
errCount.incrementAndGet();
System.out.println("database.async.command() error: " + exception.getMessage());
});
}
} catch (Exception e) {
e.printStackTrace();
}
}
@AfterEach
public void endTests() {
TestServerHelper.checkActiveDatabases();
GlobalConfiguration.resetAll();
}
} |
Beta Was this translation helpful? Give feedback.
-
|
Just updated the original test case to include both methods: |
Beta Was this translation helpful? Give feedback.
Yes, without the call to
waitCompletion(), the async thread haven't committed pending operations. This is configurable with the settingarcadedb.asyncTxBatchSizethat is se to10,240by default, namely it commits every 10K operations.You can also change that to commit every time, just use this at the beginning
GlobalConfiguration.ASYNC_TX_BATCH_SIZE.setValue(1);but it would be much slower, of course.For your use case, if I understand correctly, since you have these batches of updates, the best way is to do as with my test: one thread checks if the record exists, and then it schedules create or update in async tasks.
When everything is sent, you just wait for the finish. At that point e…