Documentation ¶
Index ¶
- Constants
- Variables
- func DecodeMapBatchInvalidationMsg(msg *proto.ClientMessage) (keys []serialization.Data, sources []types.UUID, partitions []types.UUID, ...)
- func DecodeMapInvalidationMsg(msg *proto.ClientMessage) (key serialization.Data, source types.UUID, partition types.UUID, seq int64)
- func RecomputeWithBaseTime(seconds int32) int64
- func StripBaseTime(ms int64) int32
- type DataString
- type InvalidationMetaDataFetcher
- type Manager
- type MetaDataContainer
- func (mc *MetaDataContainer) AddAndGetMissedSequenceCount(count int64) int64
- func (mc *MetaDataContainer) CASSequence(current, next int64) bool
- func (mc *MetaDataContainer) CASStaleSequence(lastKnown, lastReceived int64) bool
- func (mc *MetaDataContainer) CASUUID(prev, new types.UUID) bool
- func (mc *MetaDataContainer) MissedSequenceCount() int64
- func (mc *MetaDataContainer) ResetSequence()
- func (mc *MetaDataContainer) ResetStaleSequence()
- func (mc *MetaDataContainer) Sequence() int64
- func (mc *MetaDataContainer) SetSequence(seq int64)
- func (mc *MetaDataContainer) SetStaleSequence(seq int64)
- func (mc *MetaDataContainer) SetUUID(uuid types.UUID)
- func (mc *MetaDataContainer) StaleSequence() int64
- func (mc *MetaDataContainer) UUID() types.UUID
- type NearCache
- func (nc *NearCache) Clear()
- func (nc *NearCache) Config() *nearcache.Config
- func (nc *NearCache) Destroy()
- func (nc *NearCache) Get(key interface{}) (interface{}, bool, error)
- func (nc *NearCache) GetRecord(key interface{}) (*Record, bool)
- func (nc *NearCache) Invalidate(key interface{})
- func (nc NearCache) InvalidationRequests() int64
- func (nc NearCache) Size() int
- func (nc NearCache) Stats() nearcache.Stats
- func (nc *NearCache) TryPublishReserved(key, value interface{}, reservationID int64) (interface{}, error)
- func (nc *NearCache) TryReserveForUpdate(key interface{}, keyData serialization.Data, ups UpdateSemantic) (int64, error)
- type Record
- func (r *Record) CachedAsNil() bool
- func (r *Record) CreationTime() int64
- func (r *Record) ExpirationTime() int64
- func (r *Record) Hits() int32
- func (r *Record) IncrementHits()
- func (r *Record) InvalidationSequence() int64
- func (r *Record) IsExpiredAt(ms int64) bool
- func (r *Record) IsIdleAt(maxIdleMS, nowMS int64) bool
- func (r *Record) LastAccessTime() int64
- func (r *Record) PartitionID() int32
- func (r *Record) ReservationID() int64
- func (r *Record) SetCachedAsNil()
- func (r *Record) SetCreationTime(ms int64)
- func (r *Record) SetExpirationTIme(ms int64)
- func (r *Record) SetHits(value int32)
- func (r *Record) SetInvalidationSequence(v int64)
- func (r *Record) SetLastAccessTime(ms int64)
- func (r *Record) SetPartitionID(v int32)
- func (r *Record) SetReservationID(rid int64)
- func (r *Record) SetUUID(v types.UUID)
- func (r *Record) SetValue(value interface{})
- func (r *Record) UUID() types.UUID
- func (r *Record) Value() interface{}
- type RecordStore
- func (rs *RecordStore) Clear()
- func (rs *RecordStore) Destroy()
- func (rs *RecordStore) DoExpiration()
- func (rs *RecordStore) Get(key interface{}) (value interface{}, found bool, err error)
- func (rs *RecordStore) GetRecord(key interface{}) (*Record, bool)
- func (rs *RecordStore) Invalidate(key interface{})
- func (rs *RecordStore) InvalidationRequests() int64
- func (rs *RecordStore) Size() int
- func (rs *RecordStore) Stats() nearcache.Stats
- func (rs *RecordStore) TryPublishReserved(key, value interface{}, reservationID int64, deserialize bool) (interface{}, error)
- func (rs *RecordStore) TryReserveForUpdate(key interface{}, keyData serialization.Data, ups UpdateSemantic) (int64, error)
- type RepairingHandler
- func (h *RepairingHandler) CheckOrRepairSequence(partition int32, nextSeq int64, viaAntiEntropy bool)
- func (h *RepairingHandler) CheckOrRepairUUID(partition int32, new types.UUID)
- func (h *RepairingHandler) GetMetaDataContainer(partition int32) *MetaDataContainer
- func (h *RepairingHandler) Handle(key serialization.Data, source, partition types.UUID, seq int64) error
- func (h *RepairingHandler) HandleBatch(keys []serialization.Data, sources []types.UUID, partitions []types.UUID, ...) error
- func (h *RepairingHandler) InitSequence(partition int32, seq int64)
- func (h *RepairingHandler) InitUUID(partition int32, uuid types.UUID)
- func (h *RepairingHandler) Name() string
- func (h *RepairingHandler) UpdateLastKnownStaleSequence(md *MetaDataContainer, partition int32)
- type ReparingTask
- type StaleReadDetector
- type UpdateSemantic
Constants ¶
const ( EventIMapInvalidationMessageType = 81666 EventIMapBatchInvalidationMessageType = 81667 )
const ( EnvExpirationTaskInitialDelay = "TESTONLY_NC_EXPIRATION_INITIAL_DELAY" EnvExpirationTaskPeriod = "TESTONLY_NC_EXPIRATION_TASK_PERIOD" )
const ( RecordNotReserved int64 = -1 RecordReadPermitted int64 = -2 RecordStoreTimeNotSet int64 = -1 // see: com.hazelcast.internal.eviction.impl.strategy.sampling.SamplingEvictionStrategy#SAMPLE_COUNT RecordStoreSampleCount = 15 )
Variables ¶
var EpochTimeMillis = zeroOutMs(1514764800000)
var LFUEvictionPolicyComparator = simpleEvictionPolicyComparator{ // contains filtered or unexported fields }
var LRUEvictionPolicyComparator = simpleEvictionPolicyComparator{ // contains filtered or unexported fields }
var RandomEvictionPolicyComparator = simpleEvictionPolicyComparator{ // contains filtered or unexported fields }
Functions ¶
func DecodeMapBatchInvalidationMsg ¶
func DecodeMapBatchInvalidationMsg(msg *proto.ClientMessage) (keys []serialization.Data, sources []types.UUID, partitions []types.UUID, seqs []int64)
func DecodeMapInvalidationMsg ¶
func DecodeMapInvalidationMsg(msg *proto.ClientMessage) (key serialization.Data, source types.UUID, partition types.UUID, seq int64)
func RecomputeWithBaseTime ¶
func StripBaseTime ¶
Types ¶
type DataString ¶
type DataString string
DataString is stringified serialization.Data which is not hashable, so using this to store keys.
type InvalidationMetaDataFetcher ¶
type InvalidationMetaDataFetcher struct {
// contains filtered or unexported fields
}
InvalidationMetaDataFetcher runs on Near Cache side. An instance of this task is responsible for fetching of all Near Caches' remote metadata like last sequence numbers and partition UUIDs. port of: com.hazelcast.internal.nearcache.impl.invalidation.InvalidationMetaDataFetcher port of: com.hazelcast.client.map.impl.nearcache.invalidation.ClientMapInvalidationMetaDataFetcher
func NewInvalidationMetaDataFetcher ¶
func NewInvalidationMetaDataFetcher(cs *cluster.Service, is *invocation.Service, invFactory *cluster.ConnectionInvocationFactory, lg ilogger.LogAdaptor) InvalidationMetaDataFetcher
func (InvalidationMetaDataFetcher) Init ¶
func (df InvalidationMetaDataFetcher) Init(ctx context.Context, handler *RepairingHandler) bool
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
func (*Manager) DestroyNearCache ¶
func (*Manager) GetNearCacheStats ¶
func (*Manager) GetOrCreateNearCache ¶
func (*Manager) RepairingTask ¶
func (m *Manager) RepairingTask() *ReparingTask
type MetaDataContainer ¶
type MetaDataContainer struct {
// contains filtered or unexported fields
}
MetaDataContainer contains one partitions' invalidation metadata. port of: com.hazelcast.internal.nearcache.impl.invalidation.MetaDataContainer
func NewMetaDataContainer ¶
func NewMetaDataContainer() *MetaDataContainer
func (*MetaDataContainer) AddAndGetMissedSequenceCount ¶
func (mc *MetaDataContainer) AddAndGetMissedSequenceCount(count int64) int64
func (*MetaDataContainer) CASSequence ¶
func (mc *MetaDataContainer) CASSequence(current, next int64) bool
func (*MetaDataContainer) CASStaleSequence ¶
func (mc *MetaDataContainer) CASStaleSequence(lastKnown, lastReceived int64) bool
func (*MetaDataContainer) MissedSequenceCount ¶
func (mc *MetaDataContainer) MissedSequenceCount() int64
func (*MetaDataContainer) ResetSequence ¶
func (mc *MetaDataContainer) ResetSequence()
func (*MetaDataContainer) ResetStaleSequence ¶
func (mc *MetaDataContainer) ResetStaleSequence()
func (*MetaDataContainer) Sequence ¶
func (mc *MetaDataContainer) Sequence() int64
func (*MetaDataContainer) SetSequence ¶
func (mc *MetaDataContainer) SetSequence(seq int64)
func (*MetaDataContainer) SetStaleSequence ¶
func (mc *MetaDataContainer) SetStaleSequence(seq int64)
func (*MetaDataContainer) SetUUID ¶
func (mc *MetaDataContainer) SetUUID(uuid types.UUID)
func (*MetaDataContainer) StaleSequence ¶
func (mc *MetaDataContainer) StaleSequence() int64
func (*MetaDataContainer) UUID ¶
func (mc *MetaDataContainer) UUID() types.UUID
type NearCache ¶
type NearCache struct {
// contains filtered or unexported fields
}
func NewNearCache ¶
func NewNearCache(cfg *nearcache.Config, ss *serialization.Service, lg ilogger.LogAdaptor) *NearCache
func (*NearCache) Invalidate ¶
func (nc *NearCache) Invalidate(key interface{})
func (NearCache) InvalidationRequests ¶
InvalidationRequests returns the invalidation requests. It is used only for tests.
func (*NearCache) TryPublishReserved ¶
func (*NearCache) TryReserveForUpdate ¶
func (nc *NearCache) TryReserveForUpdate(key interface{}, keyData serialization.Data, ups UpdateSemantic) (int64, error)
type Record ¶
type Record struct {
// contains filtered or unexported fields
}
func (*Record) CachedAsNil ¶
func (*Record) CreationTime ¶
func (*Record) ExpirationTime ¶
func (*Record) IncrementHits ¶
func (r *Record) IncrementHits()
func (*Record) InvalidationSequence ¶
func (*Record) IsExpiredAt ¶
func (*Record) LastAccessTime ¶
func (*Record) PartitionID ¶
func (*Record) ReservationID ¶
func (*Record) SetCachedAsNil ¶
func (r *Record) SetCachedAsNil()
func (*Record) SetCreationTime ¶
func (*Record) SetExpirationTIme ¶
func (*Record) SetInvalidationSequence ¶
func (*Record) SetLastAccessTime ¶
func (*Record) SetPartitionID ¶
func (*Record) SetReservationID ¶
type RecordStore ¶
type RecordStore struct {
// contains filtered or unexported fields
}
func NewRecordStore ¶
func NewRecordStore(cfg *nearcache.Config, ss *serialization.Service, rc nearCacheRecordValueConverter, se nearCacheStorageEstimator) *RecordStore
func (*RecordStore) Clear ¶
func (rs *RecordStore) Clear()
func (*RecordStore) Destroy ¶
func (rs *RecordStore) Destroy()
func (*RecordStore) DoExpiration ¶
func (rs *RecordStore) DoExpiration()
func (*RecordStore) Get ¶
func (rs *RecordStore) Get(key interface{}) (value interface{}, found bool, err error)
func (*RecordStore) GetRecord ¶
func (rs *RecordStore) GetRecord(key interface{}) (*Record, bool)
func (*RecordStore) Invalidate ¶
func (rs *RecordStore) Invalidate(key interface{})
func (*RecordStore) InvalidationRequests ¶
func (rs *RecordStore) InvalidationRequests() int64
func (*RecordStore) Size ¶
func (rs *RecordStore) Size() int
func (*RecordStore) Stats ¶
func (rs *RecordStore) Stats() nearcache.Stats
func (*RecordStore) TryPublishReserved ¶
func (rs *RecordStore) TryPublishReserved(key, value interface{}, reservationID int64, deserialize bool) (interface{}, error)
func (*RecordStore) TryReserveForUpdate ¶
func (rs *RecordStore) TryReserveForUpdate(key interface{}, keyData serialization.Data, ups UpdateSemantic) (int64, error)
type RepairingHandler ¶
type RepairingHandler struct {
// contains filtered or unexported fields
}
RepairingHandler is the port of: com.hazelcast.internal.nearcache.impl.invalidation.RepairingHandler
func NewRepairingHandler ¶
func NewRepairingHandler(name string, nc *NearCache, partitionCount int32, ss *serialization.Service, ps *cluster.PartitionService, lg ilogger.LogAdaptor, uuid types.UUID) *RepairingHandler
func (*RepairingHandler) CheckOrRepairSequence ¶
func (h *RepairingHandler) CheckOrRepairSequence(partition int32, nextSeq int64, viaAntiEntropy bool)
func (*RepairingHandler) CheckOrRepairUUID ¶
func (h *RepairingHandler) CheckOrRepairUUID(partition int32, new types.UUID)
func (*RepairingHandler) GetMetaDataContainer ¶
func (h *RepairingHandler) GetMetaDataContainer(partition int32) *MetaDataContainer
func (*RepairingHandler) Handle ¶
func (h *RepairingHandler) Handle(key serialization.Data, source, partition types.UUID, seq int64) error
Handle handles a single invalidation.
func (*RepairingHandler) HandleBatch ¶
func (h *RepairingHandler) HandleBatch(keys []serialization.Data, sources []types.UUID, partitions []types.UUID, seqs []int64) error
HandleBatch handles a batch of validations.
func (*RepairingHandler) InitSequence ¶
func (h *RepairingHandler) InitSequence(partition int32, seq int64)
func (*RepairingHandler) InitUUID ¶
func (h *RepairingHandler) InitUUID(partition int32, uuid types.UUID)
func (*RepairingHandler) Name ¶
func (h *RepairingHandler) Name() string
func (*RepairingHandler) UpdateLastKnownStaleSequence ¶
func (h *RepairingHandler) UpdateLastKnownStaleSequence(md *MetaDataContainer, partition int32)
type ReparingTask ¶
type ReparingTask struct {
// contains filtered or unexported fields
}
ReparingTask runs on Near Cache side and only one instance is created per data-structure type like IMap and ICache. Repairing responsibilities of this task are:
- To scan RepairingHandlers to see if any Near Cache needs to be invalidated according to missed invalidation counts (controlled via MaxToleratedMissCount).
- To send periodic generic-operations to cluster members in order to fetch latest partition sequences and UUIDs (controlled via InvalidationMinReconciliationIntervalSeconds.
See: com.hazelcast.internal.nearcache.impl.invalidation.RepairingTask
func NewReparingTask ¶
func NewReparingTask(recInt int, maxMissCnt int, ss *serialization.Service, ps *cluster.PartitionService, lg ilogger.LogAdaptor, mf InvalidationMetaDataFetcher, uuid types.UUID, doneCh <-chan struct{}) *ReparingTask
func (*ReparingTask) DeregisterHandler ¶
func (rt *ReparingTask) DeregisterHandler(name string)
func (*ReparingTask) RegisterAndGetHandler ¶
func (rt *ReparingTask) RegisterAndGetHandler(ctx context.Context, name string, nc *NearCache) (*RepairingHandler, error)
type StaleReadDetector ¶
type StaleReadDetector struct {
// contains filtered or unexported fields
}
func NewStaleReadDetector ¶
func NewStaleReadDetector(rh *RepairingHandler, ps *icluster.PartitionService) StaleReadDetector
func (StaleReadDetector) GetMetaDataContainer ¶
func (sr StaleReadDetector) GetMetaDataContainer(partitionID int32) *MetaDataContainer
func (StaleReadDetector) GetPartitionID ¶
func (sr StaleReadDetector) GetPartitionID(keyData serialization.Data) (int32, error)
func (StaleReadDetector) IsStaleRead ¶
func (sr StaleReadDetector) IsStaleRead(rec *Record) bool
type UpdateSemantic ¶
type UpdateSemantic int8
const ( UpdateSemanticReadUpdate UpdateSemantic = iota UpdateSemanticWriteUpdate )