mirror of
https://github.com/nolen777/eagle0.git
synced 2026-07-28 20:55:44 +00:00
Instrument Postgres history append phases (#8732)
This commit is contained in:
@@ -169,6 +169,32 @@ class EagleHistoryAppendEvent extends Event {
|
||||
var succeeded: Boolean = false
|
||||
}
|
||||
|
||||
@Name("net.eagle0.eagle.HistoryAppendPhase")
|
||||
@Label("Eagle History Append Phase")
|
||||
@Category(Array("Eagle0", "History"))
|
||||
@Enabled(true)
|
||||
@StackTrace(false)
|
||||
@Threshold("0 ms")
|
||||
class EagleHistoryAppendPhaseEvent extends Event {
|
||||
@Label("Game ID")
|
||||
var gameId: String = ""
|
||||
|
||||
@Label("Phase")
|
||||
var phase: String = ""
|
||||
|
||||
@Label("Detail")
|
||||
var detail: String = ""
|
||||
|
||||
@Label("Action Result Count")
|
||||
var actionResultCount: Int = 0
|
||||
|
||||
@Label("Item Count")
|
||||
var itemCount: Long = 0L
|
||||
|
||||
@Label("Succeeded")
|
||||
var succeeded: Boolean = false
|
||||
}
|
||||
|
||||
@Name("net.eagle0.eagle.PostActionPhase")
|
||||
@Label("Eagle Post-Action Phase")
|
||||
@Category(Array("Eagle0", "Game Processing"))
|
||||
@@ -360,6 +386,28 @@ object JfrEvents {
|
||||
} else f(event)
|
||||
}
|
||||
|
||||
def historyAppendPhase[T](
|
||||
gameId: String,
|
||||
phase: String,
|
||||
actionResultCount: Int
|
||||
)(f: EagleHistoryAppendPhaseEvent => T): T = {
|
||||
val event = new EagleHistoryAppendPhaseEvent()
|
||||
event.gameId = gameId
|
||||
event.phase = phase
|
||||
event.actionResultCount = actionResultCount
|
||||
if event.isEnabled then {
|
||||
event.begin()
|
||||
try {
|
||||
val value = f(event)
|
||||
event.succeeded = true
|
||||
value
|
||||
} finally {
|
||||
event.end()
|
||||
event.commit()
|
||||
}
|
||||
} else f(event)
|
||||
}
|
||||
|
||||
def postActionPhase[T](
|
||||
gameId: => String,
|
||||
phase: String,
|
||||
|
||||
@@ -36,7 +36,7 @@ private[service] final class PerEntityBatchingConnection private (underlyingConn
|
||||
)
|
||||
.asInstanceOf[PreparedStatement]
|
||||
|
||||
def executeBatch(): Unit = statement.executeBatch(): Unit
|
||||
def executeBatch(): Array[Int] = statement.executeBatch()
|
||||
|
||||
def close(): Unit = statement.close()
|
||||
}
|
||||
@@ -61,7 +61,16 @@ private[service] final class PerEntityBatchingConnection private (underlyingConn
|
||||
)
|
||||
.asInstanceOf[Connection]
|
||||
|
||||
def executeBatches(): Unit = statementsBySql.values.foreach(_.executeBatch())
|
||||
def executeBatches(executeBatch: (String, () => Array[Int]) => Unit): Unit =
|
||||
statementsBySql.foreach {
|
||||
case (sql, batchedStatement) =>
|
||||
executeBatch(sql, () => batchedStatement.executeBatch())
|
||||
}
|
||||
|
||||
def executeBatches(): Unit =
|
||||
executeBatches { (_, executeBatch) =>
|
||||
executeBatch(): Unit
|
||||
}
|
||||
|
||||
def closeStatements(): Unit = statementsBySql.values.foreach(_.close())
|
||||
|
||||
|
||||
@@ -33,14 +33,18 @@ private[service] object PerEntityWriteDispatcher {
|
||||
* preserves the individual table writers as the single source of column binding logic, but replaces their per-entity
|
||||
* execute calls with one executeBatch per distinct statement.
|
||||
*/
|
||||
def writeAllBatch(connection: Connection, results: Vector[(Int, ActionResultT)]): Unit = {
|
||||
def writeAllBatch(
|
||||
connection: Connection,
|
||||
results: Vector[(Int, ActionResultT)],
|
||||
executeBatch: (String, () => Array[Int]) => Unit = (_, execute) => execute(): Unit
|
||||
): Unit = {
|
||||
val batchingConnection = PerEntityBatchingConnection(connection)
|
||||
try {
|
||||
results.foreach {
|
||||
case (actionSeq, actionResult) =>
|
||||
writeAll(batchingConnection.connection, actionSeq, actionResult)
|
||||
}
|
||||
batchingConnection.executeBatches()
|
||||
batchingConnection.executeBatches(executeBatch)
|
||||
} finally batchingConnection.closeStatements()
|
||||
}
|
||||
|
||||
|
||||
@@ -75,6 +75,30 @@ class PostgresHistory private[service] (
|
||||
}
|
||||
}
|
||||
|
||||
private def historyAppendPhase[A](phase: String, actionResultCount: Int)(f: => A): A =
|
||||
JfrEvents.historyAppendPhase(gameId.toHexString, phase, actionResultCount)(_ => f)
|
||||
|
||||
private def withHistoryAppendConnection[A](actionResultCount: Int)(f: => A): A =
|
||||
activeConnection match {
|
||||
case Some(_) => f
|
||||
case None =>
|
||||
val openedConnection = historyAppendPhase("connection.open", actionResultCount) {
|
||||
PostgresHistory.openConnection()
|
||||
}
|
||||
activeConnection = Some(openedConnection)
|
||||
try {
|
||||
historyAppendPhase("connection.useSchema", actionResultCount) {
|
||||
PostgresHistory.useSchema(openedConnection, schemaName)
|
||||
}
|
||||
f
|
||||
} finally {
|
||||
activeConnection = None
|
||||
historyAppendPhase("connection.close", actionResultCount) {
|
||||
PostgresHistory.closeOpenedConnection(openedConnection)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private def withLock[A](f: => A): A =
|
||||
dbLock.synchronized {
|
||||
if closed then throw new EagleInternalException(s"PostgresHistory for game ${gameId.toHexString} is closed")
|
||||
@@ -162,79 +186,124 @@ class PostgresHistory private[service] (
|
||||
|
||||
override def withNewResults(
|
||||
newResults: Vector[ActionResultWithResultingState]
|
||||
): PostgresHistory = withDbLockAfterShardokWrites {
|
||||
if newResults.isEmpty then this
|
||||
else
|
||||
JfrEvents.historyAppend(
|
||||
gameId = gameId.toHexString,
|
||||
storage = "postgres",
|
||||
startingActionCount = cachedCount,
|
||||
actionResultCount = newResults.size
|
||||
) { _ =>
|
||||
val originalAutoCommit = connection.getAutoCommit
|
||||
connection.setAutoCommit(false)
|
||||
try {
|
||||
val insertStmt = connection.prepareStatement(PostgresHistory.ActionResultInsertSql)
|
||||
try
|
||||
newResults.zipWithIndex.foreach {
|
||||
case (awrs, i) =>
|
||||
val seq = cachedCount + i
|
||||
insertStmt.setInt(1, seq)
|
||||
insertStmt.setInt(2, awrs.actionResult.actionResultType.value)
|
||||
insertStmt.setLong(3, awrs.resultingState.currentRoundId)
|
||||
val date = awrs.resultingState.currentDate
|
||||
insertStmt.setInt(4, date.year)
|
||||
insertStmt.setInt(5, date.month.value)
|
||||
awrs.actionResult.actingHeroId.fold(insertStmt.setNull(6, Types.INTEGER))(insertStmt.setInt(6, _))
|
||||
awrs.actionResult.actingFactionId.fold(insertStmt.setNull(7, Types.INTEGER))(insertStmt.setInt(7, _))
|
||||
awrs.actionResult.provinceId.fold(insertStmt.setNull(8, Types.INTEGER))(insertStmt.setInt(8, _))
|
||||
awrs.actionResult.provinceIdActed.fold(insertStmt.setNull(9, Types.INTEGER))(insertStmt.setInt(9, _))
|
||||
ActionResultEnumColumnsWriter.bindColumns(insertStmt, startPosition = 10, awrs.actionResult)
|
||||
ActionResultNumericColumnsWriter.bindColumns(insertStmt, startPosition = 14, awrs.actionResult)
|
||||
ActionResultDateBattleColumnsWriter.bindColumns(insertStmt, startPosition = 18, awrs.actionResult)
|
||||
ActionResultStructuredBlobColumnsWriter.bindColumns(insertStmt, startPosition = 21, awrs.actionResult)
|
||||
insertStmt.addBatch()
|
||||
}
|
||||
insertStmt.executeBatch(): Unit
|
||||
finally insertStmt.close()
|
||||
end try
|
||||
): PostgresHistory = {
|
||||
flushShardokWrites()
|
||||
withLock {
|
||||
if newResults.isEmpty then withConnection(this)
|
||||
else {
|
||||
val actionResultCount = newResults.size
|
||||
JfrEvents.historyAppend(
|
||||
gameId = gameId.toHexString,
|
||||
storage = "postgres",
|
||||
startingActionCount = cachedCount,
|
||||
actionResultCount = actionResultCount
|
||||
) { _ =>
|
||||
withHistoryAppendConnection(actionResultCount) {
|
||||
appendNewResults(newResults)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
PerEntityWriteDispatcher.writeAllBatch(
|
||||
connection,
|
||||
newResults.zipWithIndex.map { case (awrs, i) => cachedCount + i -> awrs.actionResult }
|
||||
)
|
||||
|
||||
val snapshotStmt =
|
||||
connection.prepareStatement("INSERT INTO state_snapshots (boundary_action_seq, payload) VALUES (?, ?)")
|
||||
try
|
||||
newResults.zipWithIndex.foreach {
|
||||
case (awrs, i) =>
|
||||
val boundary = cachedCount + i + 1
|
||||
if boundary % SnapshotInterval == 0 then {
|
||||
snapshotStmt.setInt(1, boundary)
|
||||
snapshotStmt.setBytes(2, GameStateConverter.toProto(awrs.resultingState).toByteArray)
|
||||
snapshotStmt.addBatch(): Unit
|
||||
}
|
||||
}
|
||||
snapshotStmt.executeBatch(): Unit
|
||||
finally snapshotStmt.close()
|
||||
|
||||
connection.commit()
|
||||
private def appendNewResults(newResults: Vector[ActionResultWithResultingState]): PostgresHistory = {
|
||||
val actionResultCount = newResults.size
|
||||
val originalAutoCommit = historyAppendPhase("transaction.begin", actionResultCount) {
|
||||
val original = connection.getAutoCommit
|
||||
connection.setAutoCommit(false)
|
||||
original
|
||||
}
|
||||
try {
|
||||
JfrEvents.historyAppendPhase(gameId.toHexString, "actionResults.batch", actionResultCount) { event =>
|
||||
event.itemCount = actionResultCount
|
||||
val insertStmt = connection.prepareStatement(PostgresHistory.ActionResultInsertSql)
|
||||
try
|
||||
newResults.zipWithIndex.foreach {
|
||||
case (awrs, i) =>
|
||||
val seq = cachedCount + i
|
||||
actionCache(seq) = awrs.actionResult
|
||||
stateCache(seq + 1) = awrs.resultingState
|
||||
val seq = cachedCount + i
|
||||
insertStmt.setInt(1, seq)
|
||||
insertStmt.setInt(2, awrs.actionResult.actionResultType.value)
|
||||
insertStmt.setLong(3, awrs.resultingState.currentRoundId)
|
||||
val date = awrs.resultingState.currentDate
|
||||
insertStmt.setInt(4, date.year)
|
||||
insertStmt.setInt(5, date.month.value)
|
||||
awrs.actionResult.actingHeroId.fold(insertStmt.setNull(6, Types.INTEGER))(insertStmt.setInt(6, _))
|
||||
awrs.actionResult.actingFactionId.fold(insertStmt.setNull(7, Types.INTEGER))(insertStmt.setInt(7, _))
|
||||
awrs.actionResult.provinceId.fold(insertStmt.setNull(8, Types.INTEGER))(insertStmt.setInt(8, _))
|
||||
awrs.actionResult.provinceIdActed.fold(insertStmt.setNull(9, Types.INTEGER))(insertStmt.setInt(9, _))
|
||||
ActionResultEnumColumnsWriter.bindColumns(insertStmt, startPosition = 10, awrs.actionResult)
|
||||
ActionResultNumericColumnsWriter.bindColumns(insertStmt, startPosition = 14, awrs.actionResult)
|
||||
ActionResultDateBattleColumnsWriter.bindColumns(insertStmt, startPosition = 18, awrs.actionResult)
|
||||
ActionResultStructuredBlobColumnsWriter.bindColumns(insertStmt, startPosition = 21, awrs.actionResult)
|
||||
insertStmt.addBatch()
|
||||
}
|
||||
cachedCount += newResults.length
|
||||
} catch {
|
||||
case e: Exception =>
|
||||
connection.rollback()
|
||||
throw e
|
||||
} finally connection.setAutoCommit(originalAutoCommit)
|
||||
insertStmt.executeBatch(): Unit
|
||||
finally insertStmt.close()
|
||||
end try
|
||||
this
|
||||
}
|
||||
|
||||
JfrEvents.historyAppendPhase(gameId.toHexString, "entities.allBatches", actionResultCount) { event =>
|
||||
var executedBatchCount = 0L
|
||||
PerEntityWriteDispatcher.writeAllBatch(
|
||||
connection,
|
||||
newResults.zipWithIndex.map { case (awrs, i) => cachedCount + i -> awrs.actionResult },
|
||||
executeBatch = (sql, execute) => {
|
||||
JfrEvents.historyAppendPhase(gameId.toHexString, "entities.batch", actionResultCount) { batchEvent =>
|
||||
batchEvent.detail = sql
|
||||
val results = execute()
|
||||
batchEvent.itemCount = results.length
|
||||
}
|
||||
executedBatchCount += 1
|
||||
}
|
||||
)
|
||||
event.itemCount = executedBatchCount
|
||||
}
|
||||
|
||||
JfrEvents.historyAppendPhase(gameId.toHexString, "snapshots.batch", actionResultCount) { event =>
|
||||
val snapshotStmt =
|
||||
connection.prepareStatement("INSERT INTO state_snapshots (boundary_action_seq, payload) VALUES (?, ?)")
|
||||
try {
|
||||
var snapshotCount = 0L
|
||||
newResults.zipWithIndex.foreach {
|
||||
case (awrs, i) =>
|
||||
val boundary = cachedCount + i + 1
|
||||
if boundary % SnapshotInterval == 0 then {
|
||||
snapshotStmt.setInt(1, boundary)
|
||||
snapshotStmt.setBytes(2, GameStateConverter.toProto(awrs.resultingState).toByteArray)
|
||||
snapshotStmt.addBatch(): Unit
|
||||
snapshotCount += 1
|
||||
}
|
||||
}
|
||||
snapshotStmt.executeBatch(): Unit
|
||||
event.itemCount = snapshotCount
|
||||
} finally snapshotStmt.close()
|
||||
end try
|
||||
}
|
||||
|
||||
historyAppendPhase("transaction.commit", actionResultCount) {
|
||||
connection.commit()
|
||||
}
|
||||
historyAppendPhase("cache.update", actionResultCount) {
|
||||
newResults.zipWithIndex.foreach {
|
||||
case (awrs, i) =>
|
||||
val seq = cachedCount + i
|
||||
actionCache(seq) = awrs.actionResult
|
||||
stateCache(seq + 1) = awrs.resultingState
|
||||
}
|
||||
cachedCount += newResults.length
|
||||
}
|
||||
} catch {
|
||||
case e: Exception =>
|
||||
historyAppendPhase("transaction.rollback", actionResultCount) {
|
||||
connection.rollback()
|
||||
}
|
||||
throw e
|
||||
} finally
|
||||
historyAppendPhase("transaction.restoreAutoCommit", actionResultCount) {
|
||||
connection.setAutoCommit(originalAutoCommit)
|
||||
}
|
||||
end try
|
||||
this
|
||||
}
|
||||
|
||||
override def truncateTo(targetActionCount: Int): PostgresHistory = withDbLockAfterShardokWrites {
|
||||
|
||||
@@ -61,4 +61,39 @@ class JfrEventsTest extends AnyFlatSpec with Matchers {
|
||||
|
||||
metadataEvaluated shouldBe false
|
||||
}
|
||||
|
||||
"historyAppendPhase" should "record phase metadata and counts" in {
|
||||
val recordingPath = Files.createTempFile("eagle-history-append-phase", ".jfr")
|
||||
val recording = new Recording()
|
||||
try {
|
||||
recording.enable("net.eagle0.eagle.HistoryAppendPhase").withThreshold(Duration.ZERO)
|
||||
recording.start()
|
||||
|
||||
val result = JfrEvents.historyAppendPhase("game-456", "entities.batch", actionResultCount = 4) { event =>
|
||||
event.detail = "INSERT INTO action_changed_heroes"
|
||||
event.itemCount = 9
|
||||
"result"
|
||||
}
|
||||
|
||||
recording.stop()
|
||||
recording.dump(recordingPath)
|
||||
|
||||
result shouldBe "result"
|
||||
val event = RecordingFile
|
||||
.readAllEvents(recordingPath)
|
||||
.asScala
|
||||
.find(_.getEventType.getName == "net.eagle0.eagle.HistoryAppendPhase")
|
||||
.getOrElse(fail("HistoryAppendPhase event was not recorded"))
|
||||
event.getString("gameId") shouldBe "game-456"
|
||||
event.getString("phase") shouldBe "entities.batch"
|
||||
event.getString("detail") shouldBe "INSERT INTO action_changed_heroes"
|
||||
event.getInt("actionResultCount") shouldBe 4
|
||||
event.getLong("itemCount") shouldBe 9L
|
||||
event.getBoolean("succeeded") shouldBe true
|
||||
event.getStackTrace shouldBe null
|
||||
} finally {
|
||||
recording.close()
|
||||
Files.deleteIfExists(recordingPath): Unit
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,7 +31,14 @@ class PerEntityBatchingConnectionTest extends AnyFlatSpec with Matchers with Moc
|
||||
second.executeUpdate() shouldBe 1
|
||||
second.close()
|
||||
|
||||
batching.executeBatches()
|
||||
var observedSql = Option.empty[String]
|
||||
var observedResults = Vector.empty[Int]
|
||||
batching.executeBatches { (batchSql, executeBatch) =>
|
||||
observedSql = Some(batchSql)
|
||||
observedResults = executeBatch().toVector
|
||||
}
|
||||
observedSql shouldBe Some(sql)
|
||||
observedResults shouldBe Vector(1, 1)
|
||||
batching.closeStatements()
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user