Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Keep the shard ids sorted for logging and api #147

Merged
merged 6 commits into from
Oct 17, 2024
Merged

Keep the shard ids sorted for logging and api #147

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d803368
Use SortedMap for in-memory shard state
yoohaemin Oct 16, 2024
7012896
Use BitSet for assignments and unassignments
yoohaemin Oct 16, 2024
782226f
Specify SortedMap for shards
yoohaemin Oct 16, 2024
40c950e
Revert and sort on log
yoohaemin Oct 16, 2024
3d0c318
Deduplicate sorting of shards, use implementation that works on 2.12
yoohaemin Oct 17, 2024
b676da0
Format GraphQLApi file
yoohaemin Oct 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions core/src/main/scala/com/devsisters/shardcake/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,14 @@
package com.devsisters

import scala.collection.mutable

package object shardcake {
type ShardId = Int
type EpochMillis = Long

private[shardcake] def renderShardIds(ids: Iterable[ShardId]): String =
ids
.foldLeft(mutable.BitSet.empty)(_ += _)
.mkString("[", ", ", "]")

}
6 changes: 3 additions & 3 deletions entities/src/main/scala/com/devsisters/shardcake/Sharding.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,23 +102,23 @@ class Sharding private (
shardAssignments.update(shards.foldLeft(_) { case (map, shard) => map.updated(shard, address) }) *>
Metrics.shards.incrementBy(shards.size) *>
startSingletonsIfNeeded *>
ZIO.logDebug(s"Assigned shards: $shards")
ZIO.logDebug(s"Assigned shards: ${renderShardIds(shards)}")
}
.unit

private[shardcake] def unassign(shards: Set[ShardId]): UIO[Unit] =
shardAssignments.update(shards.foldLeft(_) { case (map, shard) =>
if (map.get(shard).contains(address)) map - shard else map
}) *>
ZIO.logDebug(s"Unassigning shards: $shards") *>
ZIO.logDebug(s"Unassigning shards: ${renderShardIds(shards)}") *>
entityStates.get.flatMap(state =>
ZIO.foreachDiscard(state.values)(
_.entityManager.terminateEntitiesOnShards(shards) // this will return once all shards are terminated
)
) *>
Metrics.shards.decrementBy(shards.size) *>
stopSingletonsIfNeeded *>
ZIO.logDebug(s"Unassigned shards: $shards")
ZIO.logDebug(s"Unassigned shards: ${renderShardIds(shards)}")

private[shardcake] def isEntityOnLocalShards(recipientType: RecipientType[_], entityId: String): UIO[Boolean] =
for {
Expand Down
6 changes: 5 additions & 1 deletion manager/src/main/scala/com/devsisters/shardcake/GraphQLApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,11 @@ object GraphQLApi extends GenericSchema[ShardManager] {
val api: GraphQL[ShardManager] =
graphQL[ShardManager, Queries, Mutations, Subscriptions](
RootResolver(
Queries(ZIO.serviceWithZIO(_.getAssignments.map(_.map { case (k, v) => Assignment(k, v) }.toList))),
Queries(
ZIO.serviceWithZIO(
_.getAssignments.map(_.map { case (k, v) => Assignment(k, v) }.toList.sortBy(_.shardId))
)
),
Mutations(
pod => ZIO.serviceWithZIO(_.register(pod)),
pod => ZIO.serviceWithZIO(_.unregister(pod.address)),
Expand Down
8 changes: 6 additions & 2 deletions manager/src/main/scala/com/devsisters/shardcake/ShardManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -317,8 +317,12 @@ object ShardManager {

sealed trait ShardingEvent
object ShardingEvent {
case class ShardsAssigned(pod: PodAddress, shards: Set[ShardId]) extends ShardingEvent
case class ShardsUnassigned(pod: PodAddress, shards: Set[ShardId]) extends ShardingEvent
case class ShardsAssigned(pod: PodAddress, shards: Set[ShardId]) extends ShardingEvent {
override def toString: String = s"ShardsAssigned(pod=$pod, shards=${renderShardIds(shards)})"
}
case class ShardsUnassigned(pod: PodAddress, shards: Set[ShardId]) extends ShardingEvent {
override def toString: String = s"ShardsUnassigned(pod=$pod, shards=${renderShardIds(shards)})"
}
case class PodRegistered(pod: PodAddress) extends ShardingEvent
case class PodUnregistered(pod: PodAddress) extends ShardingEvent
case class PodHealthChecked(pod: PodAddress) extends ShardingEvent
Expand Down
Loading