Further parallelize game asset loading

src/main/kotlin/ru/dbotthepony/kstarbound/Ext.kt

@@ -7,7 +7,12 @@ import com.google.gson.TypeAdapter
 import com.google.gson.TypeAdapterFactory
 import com.google.gson.reflect.TypeToken
 import com.google.gson.stream.JsonReader
+import ru.dbotthepony.kstarbound.api.IStarboundFile
+import ru.dbotthepony.kstarbound.util.KOptional
 import java.util.Arrays
+import java.util.concurrent.Callable
+import java.util.concurrent.ForkJoinPool
+import java.util.concurrent.ForkJoinTask
 import java.util.stream.Stream
 import kotlin.reflect.KProperty
 import kotlin.reflect.full.createType
@@ -46,3 +51,34 @@ fun String.sintern(): String = Starbound.strings.intern(this)
 
 inline fun <reified T> Gson.fromJson(reader: JsonReader): T? = fromJson<T>(reader, T::class.java)
 
+fun <T : Any> Collection<IStarboundFile>.batch(executor: ForkJoinPool, batchSize: Int = 16, mapper: (IStarboundFile) -> KOptional<RegistryObject<T>>): Stream<RegistryObject<T>> {
+	require(batchSize >= 1) { "Invalid batch size: $batchSize" }
+
+	if (batchSize == 1 || size <= batchSize) {
+		val tasks = ArrayList<ForkJoinTask<KOptional<RegistryObject<T>>>>()
+
+		for (listedFile in this) {
+			tasks.add(executor.submit(Callable { mapper.invoke(listedFile) }))
+		}
+
+		return tasks.stream().map { it.join() }.filter { it.isPresent }.map { it.value }
+	}
+
+	val batches = ArrayList<ForkJoinTask<List<KOptional<RegistryObject<T>>>>>()
+	var batch = ArrayList<IStarboundFile>(batchSize)
+
+	for (listedFile in this) {
+		batch.add(listedFile)
+
+		if (batch.size >= batchSize) {
+			val mbatch = batch
+			batches.add(executor.submit(Callable { mbatch.map { mapper.invoke(it) } }))
+			batch = ArrayList(batchSize)
+		}
+	}
+
+	if (batch.isNotEmpty())
+		batches.add(executor.submit(Callable { batch.map { mapper.invoke(it) } }))
+
+	return batches.stream().flatMap { it.join().stream() }.filter { it.isPresent }.map { it.value }
+}

src/main/kotlin/ru/dbotthepony/kstarbound/LoadingLog.kt

@@ -2,6 +2,7 @@ package ru.dbotthepony.kstarbound
 
 import it.unimi.dsi.fastutil.ints.IntIterators
 import it.unimi.dsi.fastutil.objects.ObjectIterators
+import java.util.concurrent.atomic.AtomicInteger
 
 interface ILoadingLog : Iterable<ILoadingLog.ILine> {
 	interface ILine {
@@ -9,7 +10,7 @@ interface ILoadingLog : Iterable<ILoadingLog.ILine> {
 		val progress: Float
 			get() = if (maxElements > 0) elements.toFloat() / maxElements.toFloat() else 0f
 
-		var elements: Int
+		val elements: AtomicInteger
 		var maxElements: Int
 	}
 
@@ -20,9 +21,8 @@ interface ILoadingLog : Iterable<ILoadingLog.ILine> {
 			get() = ""
 			set(value) {}
 
-		override var elements: Int
+		override var elements: AtomicInteger = AtomicInteger()
-			get() = 0
+
-			set(value) {}
 		override var maxElements: Int
 			get() = 0
 			set(value) {}
@@ -39,6 +39,7 @@ interface ILoadingLog : Iterable<ILoadingLog.ILine> {
 
 class LoadingLog : ILoadingLog {
 	private val lines = arrayOfNulls<Line>(128)
+	@Volatile
 	private var index = 0
 	private val lock = Any()
 	private var size = 0
@@ -56,8 +57,7 @@ class LoadingLog : ILoadingLog {
 				lastActivity = System.nanoTime()
 			}
 
-		@Volatile
+		override val elements = AtomicInteger()
-		override var elements: Int = 0
 		override var maxElements: Int = 0
 
 		init {

src/main/kotlin/ru/dbotthepony/kstarbound/ObjectRegistry.kt

@@ -1,13 +1,10 @@
 package ru.dbotthepony.kstarbound
 
-import com.google.gson.Gson
 import com.google.gson.JsonArray
 import com.google.gson.JsonElement
 import com.google.gson.JsonObject
 import com.google.gson.JsonPrimitive
 import com.google.gson.internal.bind.JsonTreeReader
-import it.unimi.dsi.fastutil.ints.Int2ObjectMap
-import it.unimi.dsi.fastutil.ints.Int2ObjectMaps
 import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap
 import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap
 import org.apache.logging.log4j.LogManager
@@ -17,7 +14,6 @@ import ru.dbotthepony.kstarbound.util.AssetPathStack
 import ru.dbotthepony.kstarbound.util.set
 import ru.dbotthepony.kstarbound.util.traverseJsonPath
 import java.util.*
-import java.util.concurrent.atomic.AtomicInteger
 import kotlin.reflect.KClass
 
 inline fun <reified T : Any> ObjectRegistry(name: String, noinline key: ((T) -> String)? = null, noinline intKey: ((T) -> Int)? = null): ObjectRegistry<T> {
@@ -147,6 +143,10 @@ class ObjectRegistry<T : Any>(val clazz: KClass<T>, val name: String, val key: (
 		}
 	}
 
+	fun add(value: RegistryObject<T>): Boolean {
+		return add(value, this.key?.invoke(value.value) ?: throw UnsupportedOperationException("No key mapper"))
+	}
+
 	fun add(value: T, json: JsonElement, file: IStarboundFile): Boolean {
 		return add(RegistryObject(value, json, file), this.key?.invoke(value) ?: throw UnsupportedOperationException("No key mapper"))
 	}

src/main/kotlin/ru/dbotthepony/kstarbound/RecipeRegistry.kt

@@ -9,8 +9,10 @@ import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap
 import org.apache.logging.log4j.LogManager
 import ru.dbotthepony.kstarbound.api.IStarboundFile
 import ru.dbotthepony.kstarbound.defs.player.RecipeDefinition
+import ru.dbotthepony.kstarbound.util.KOptional
 import java.util.Collections
 import java.util.LinkedList
+import java.util.concurrent.Callable
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.ForkJoinPool
 import java.util.concurrent.ForkJoinTask
@@ -62,29 +64,29 @@ object RecipeRegistry {
 	fun load(log: ILoadingLog, fileTree: Map<String, List<IStarboundFile>>, executor: ForkJoinPool): List<ForkJoinTask<*>> {
 		val files = fileTree["recipe"] ?: return emptyList()
 
-		return listOf(executor.submit(Runnable {
+		val elements = Starbound.gson.getAdapter(JsonElement::class.java)
+		val recipes = Starbound.gson.getAdapter(RecipeDefinition::class.java)
+
+		return listOf(executor.submit {
 			val line = log.line("Loading recipes...")
 			val time = System.nanoTime()
 			line.maxElements = files.size
 
-			for ((i, listedFile) in files.withIndex()) {
+			files.batch(executor) { listedFile ->
 				try {
 					line.text = ("Loading $listedFile")
-					val json = Starbound.gson.fromJson(listedFile.reader(), JsonElement::class.java)
+					val json = elements.read(listedFile.jsonReader())
-					val value = Starbound.gson.fromJson<RecipeDefinition>(JsonTreeReader(json), RecipeDefinition::class.java)
+					val value = recipes.fromJsonTree(json)
-					add(RegistryObject(value, json, listedFile))
+					line.elements.incrementAndGet()
-					line.elements++
+					KOptional(RegistryObject(value, json, listedFile))
 				} catch (err: Throwable) {
 					LOGGER.error("Loading recipe definition file $listedFile", err)
-					line.elements++
+					line.elements.incrementAndGet()
+					KOptional.empty()
 				}
-
+			}.forEach { add(it) }
-				if (Starbound.terminateLoading) {
-					return@Runnable
-				}
-			}
 
 			line.text = "Loaded recipes in ${((System.nanoTime() - time) / 1_000_000.0).toLong()}ms"
-		}))
+		})
 	}
 }

src/main/kotlin/ru/dbotthepony/kstarbound/Registries.kt

@@ -1,5 +1,6 @@
 package ru.dbotthepony.kstarbound
 
+import com.google.gson.JsonElement
 import com.google.gson.JsonObject
 import com.google.gson.internal.bind.JsonTreeReader
 import com.google.gson.stream.JsonReader
@@ -34,6 +35,8 @@ import ru.dbotthepony.kstarbound.defs.tile.LiquidDefinition
 import ru.dbotthepony.kstarbound.defs.tile.MaterialModifier
 import ru.dbotthepony.kstarbound.defs.tile.TileDefinition
 import ru.dbotthepony.kstarbound.util.AssetPathStack
+import ru.dbotthepony.kstarbound.util.KOptional
+import java.util.concurrent.Callable
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.ForkJoinPool
 import java.util.concurrent.ForkJoinTask
@@ -75,29 +78,37 @@ object Registries {
 		line.text = ("Loaded $name in ${System.currentTimeMillis() - time}ms".also(LOGGER::info))
 	}
 
-	private fun <T : Any> loadStage(
+	private inline fun <reified T : Any> loadStage(
 		log: ILoadingLog,
+		executor: ForkJoinPool,
 		registry: ObjectRegistry<T>,
 		files: List<IStarboundFile>,
+		name: String = registry.name
 	) {
+		val adapter = Starbound.gson.getAdapter(T::class.java)
+		val elementAdapter = Starbound.gson.getAdapter(JsonElement::class.java)
+
 		loadStage(log, loader = {
 			it.maxElements = files.size
 
-			for (listedFile in files) {
+			files.batch(executor) { listedFile ->
 				try {
 					it.text = "Loading $listedFile"
-					registry.add(listedFile)
+
-					it.elements++
+					val result = AssetPathStack(listedFile.computeDirectory()) {
+						val elem = elementAdapter.read(listedFile.jsonReader())
+						RegistryObject(adapter.fromJsonTree(elem), elem, listedFile)
+					}
+
+					it.elements.incrementAndGet()
+					KOptional(result)
 				} catch (err: Throwable) {
 					LOGGER.error("Loading ${registry.name} definition file $listedFile", err)
-					it.elements++
+					it.elements.incrementAndGet()
+					KOptional.empty()
 				}
-
+			}.forEach { registry.add(it) }
-				if (Starbound.terminateLoading) {
+		}, name)
-					break
-				}
-			}
-		}, registry.name)
 	}
 
 	fun load(log: ILoadingLog, fileTree: Map<String, List<IStarboundFile>>, executor: ForkJoinPool): List<ForkJoinTask<*>> {
@@ -109,19 +120,19 @@ object Registries {
 		tasks.add(executor.submit { loadStage(log, { loadJson2Functions(it, fileTree["2functions"] ?: listOf()) }, "json 2functions") })
 		tasks.add(executor.submit { loadStage(log, { loadTreasurePools(it, fileTree["treasurepools"] ?: listOf()) }, "treasure pools") })
 
-		tasks.add(executor.submit { loadStage(log, tiles, fileTree["material"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, tiles, fileTree["material"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, tileModifiers, fileTree["matmod"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, tileModifiers, fileTree["matmod"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, liquid, fileTree["liquid"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, liquid, fileTree["liquid"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, worldObjects, fileTree["object"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, worldObjects, fileTree["object"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, statusEffects, fileTree["statuseffect"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, statusEffects, fileTree["statuseffect"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, species, fileTree["species"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, species, fileTree["species"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, particles, fileTree["particle"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, particles, fileTree["particle"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, questTemplates, fileTree["questtemplate"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, questTemplates, fileTree["questtemplate"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, techs, fileTree["tech"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, techs, fileTree["tech"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, npcTypes, fileTree["npctype"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, npcTypes, fileTree["npctype"] ?: listOf()) })
-		// tasks.add(executor.submit { loadStage(log, projectiles, ext2files["projectile"] ?: listOf()) })
+		// tasks.add(executor.submit { loadStage(log, executor, projectiles, ext2files["projectile"] ?: listOf()) })
-		// tasks.add(executor.submit { loadStage(log, tenants, ext2files["tenant"] ?: listOf()) })
+		// tasks.add(executor.submit { loadStage(log, executor, tenants, ext2files["tenant"] ?: listOf()) })
-		tasks.add(executor.submit { loadStage(log, monsterSkills, fileTree["monsterskill"] ?: listOf()) })
+		tasks.add(executor.submit { loadStage(log, executor, monsterSkills, fileTree["monsterskill"] ?: listOf()) })
 		// tasks.add(executor.submit { loadStage(log, _monsterTypes, ext2files["monstertype"] ?: listOf()) })
 
 		return tasks
@@ -142,31 +153,30 @@ object Registries {
 		)
 
 		val tasks = ArrayList<ForkJoinTask<*>>()
+		val objects = Starbound.gson.getAdapter(JsonObject::class.java)
 
 		for ((ext, clazz) in fileMap) {
 			val fileList = files[ext] ?: continue
+			val adapter = Starbound.gson.getAdapter(clazz)
 
 			tasks.add(executor.submit {
 				val line = log.line("Loading items '$ext'")
-				val time = System.nanoTime()
 				line.maxElements = fileList.size
+				val time = System.nanoTime()
 
-				for (listedFile in fileList) {
+				fileList.batch(executor) { listedFile ->
 					try {
 						line.text = "Loading $listedFile"
-						val json = Starbound.gson.fromJson(listedFile.reader(), JsonObject::class.java)
+						val json = objects.read(listedFile.jsonReader())
-						val def: IItemDefinition = AssetPathStack(listedFile.computeDirectory()) { Starbound.gson.fromJson(JsonTreeReader(json), clazz) }
+						val def = AssetPathStack(listedFile.computeDirectory()) { adapter.fromJsonTree(json) }
-						items.add(def, json, listedFile)
+						line.elements.incrementAndGet()
-						line.elements++
+						KOptional(RegistryObject(def, json, listedFile))
 					} catch (err: Throwable) {
 						LOGGER.error("Loading item definition file $listedFile", err)
-						line.elements++
+						line.elements.incrementAndGet()
+						KOptional.empty()
 					}
-
+				}.forEach { items.add(it) }
-					if (Starbound.terminateLoading) {
-						return@submit
-					}
-				}
 
 				line.text = "Loaded items '$ext' in ${((System.nanoTime() - time) / 1_000_000.0).toLong()}ms"
 			})

src/main/kotlin/ru/dbotthepony/kstarbound/io/json/builder/FactoryAdapter.kt

@@ -33,6 +33,7 @@ import ru.dbotthepony.kstarbound.io.json.consumeNull
 import ru.dbotthepony.kstarbound.io.json.value
 import ru.dbotthepony.kstarbound.util.Either
 import java.lang.reflect.Constructor
+import java.util.Collections
 import java.util.function.Function
 import kotlin.jvm.internal.DefaultConstructorMarker
 import kotlin.properties.Delegates
@@ -56,7 +57,7 @@ class FactoryAdapter<T : Any> private constructor(
 	private val elements: TypeAdapter<JsonElement>
 ) : TypeAdapter<T>() {
 	private val name2index = Object2ObjectArrayMap<String, IntArrayList>()
-	private val loggedMisses = ObjectArraySet<String>()
+	private val loggedMisses = Collections.synchronizedSet(ObjectArraySet<String>())
 
 	init {
 		if (asJsonArray && types.any { it.isFlat }) {