| /* |
| * 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 org.apache.ignite.internal.processors.cache.binary; |
| |
| import java.io.File; |
| import java.io.Serializable; |
| import java.math.BigDecimal; |
| import java.nio.ByteBuffer; |
| import java.nio.file.Paths; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import javax.cache.CacheException; |
| import org.apache.ignite.IgniteBinary; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteClientDisconnectedException; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.IgniteLogger; |
| import org.apache.ignite.IgniteSystemProperties; |
| import org.apache.ignite.binary.BinaryField; |
| import org.apache.ignite.binary.BinaryObject; |
| import org.apache.ignite.binary.BinaryObjectBuilder; |
| import org.apache.ignite.binary.BinaryObjectException; |
| import org.apache.ignite.binary.BinaryType; |
| import org.apache.ignite.binary.BinaryTypeConfiguration; |
| import org.apache.ignite.cache.affinity.AffinityKeyMapper; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.BinaryConfiguration; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.DataStorageConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.internal.GridKernalContext; |
| import org.apache.ignite.internal.IgniteFeatures; |
| import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.IgniteNodeAttributes; |
| import org.apache.ignite.internal.NodeStoppingException; |
| import org.apache.ignite.internal.UnregisteredBinaryTypeException; |
| import org.apache.ignite.internal.binary.BinaryContext; |
| import org.apache.ignite.internal.binary.BinaryEnumObjectImpl; |
| import org.apache.ignite.internal.binary.BinaryFieldMetadata; |
| import org.apache.ignite.internal.binary.BinaryMarshaller; |
| import org.apache.ignite.internal.binary.BinaryMetadata; |
| import org.apache.ignite.internal.binary.BinaryMetadataHandler; |
| import org.apache.ignite.internal.binary.BinaryObjectEx; |
| import org.apache.ignite.internal.binary.BinaryObjectImpl; |
| import org.apache.ignite.internal.binary.BinaryObjectOffheapImpl; |
| import org.apache.ignite.internal.binary.BinaryTypeImpl; |
| import org.apache.ignite.internal.binary.BinaryUtils; |
| import org.apache.ignite.internal.binary.GridBinaryMarshaller; |
| import org.apache.ignite.internal.binary.builder.BinaryObjectBuilderImpl; |
| import org.apache.ignite.internal.binary.streams.BinaryInputStream; |
| import org.apache.ignite.internal.binary.streams.BinaryOffheapInputStream; |
| import org.apache.ignite.internal.processors.GridProcessorAdapter; |
| import org.apache.ignite.internal.processors.cache.CacheDefaultBinaryAffinityKeyMapper; |
| import org.apache.ignite.internal.processors.cache.CacheObject; |
| import org.apache.ignite.internal.processors.cache.CacheObjectByteArrayImpl; |
| import org.apache.ignite.internal.processors.cache.CacheObjectContext; |
| import org.apache.ignite.internal.processors.cache.CacheObjectImpl; |
| import org.apache.ignite.internal.processors.cache.CacheObjectValueContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheDefaultAffinityKeyMapper; |
| import org.apache.ignite.internal.processors.cache.GridCacheUtils; |
| import org.apache.ignite.internal.processors.cache.IncompleteCacheObject; |
| import org.apache.ignite.internal.processors.cache.KeyCacheObject; |
| import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx; |
| import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor; |
| import org.apache.ignite.internal.processors.cacheobject.UserCacheObjectByteArrayImpl; |
| import org.apache.ignite.internal.processors.cacheobject.UserCacheObjectImpl; |
| import org.apache.ignite.internal.processors.cacheobject.UserKeyCacheObjectImpl; |
| import org.apache.ignite.internal.processors.query.QueryUtils; |
| import org.apache.ignite.internal.util.GridUnsafe; |
| import org.apache.ignite.internal.util.IgniteUtils; |
| import org.apache.ignite.internal.util.MutableSingletonList; |
| import org.apache.ignite.internal.util.future.GridFutureAdapter; |
| import org.apache.ignite.internal.util.lang.GridMapEntry; |
| import org.apache.ignite.internal.util.tostring.GridToStringExclude; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.T1; |
| import org.apache.ignite.internal.util.typedef.T2; |
| import org.apache.ignite.internal.util.typedef.internal.A; |
| import org.apache.ignite.internal.util.typedef.internal.CU; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.lang.IgniteClosure; |
| import org.apache.ignite.lang.IgniteFuture; |
| import org.apache.ignite.lang.IgniteUuid; |
| import org.apache.ignite.marshaller.Marshaller; |
| import org.apache.ignite.spi.IgniteNodeValidationResult; |
| import org.apache.ignite.spi.discovery.DiscoveryDataBag; |
| import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData; |
| import org.apache.ignite.spi.discovery.IgniteDiscoveryThread; |
| import org.apache.ignite.thread.IgniteThread; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static java.util.concurrent.TimeUnit.NANOSECONDS; |
| import static org.apache.ignite.IgniteSystemProperties.IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK; |
| import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAIT_SCHEMA_UPDATE; |
| import static org.apache.ignite.IgniteSystemProperties.getBoolean; |
| import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.BINARY_PROC; |
| import static org.apache.ignite.internal.binary.BinaryUtils.mergeMetadata; |
| |
| /** |
| * Binary processor implementation. |
| */ |
| public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter implements IgniteCacheObjectProcessor { |
| /** Immutable classes. */ |
| private static final Collection<Class<?>> IMMUTABLE_CLS = new HashSet<>(); |
| |
| /** */ |
| private volatile boolean discoveryStarted; |
| |
| /** */ |
| private volatile IgniteFuture<?> reconnectFut; |
| |
| /** */ |
| private BinaryContext binaryCtx; |
| |
| /** */ |
| private Marshaller marsh; |
| |
| /** */ |
| private GridBinaryMarshaller binaryMarsh; |
| |
| /** */ |
| private BinaryMetadataFileStore metadataFileStore; |
| |
| /** |
| * Custom folder specifying local folder for {@link #metadataFileStore}.<br> |
| * {@code null} means no specific folder is configured. <br> |
| * In this case folder for metadata is composed from work directory and consistentId <br> |
| */ |
| @Nullable private File binaryMetadataFileStoreDir; |
| |
| /** How long to wait for schema if no updates in progress. */ |
| private long waitSchemaTimeout = IgniteSystemProperties.getLong(IGNITE_WAIT_SCHEMA_UPDATE, 30_000); |
| |
| /** For tests. */ |
| @SuppressWarnings("PublicField") |
| public static boolean useTestBinaryCtx; |
| |
| /** */ |
| @GridToStringExclude |
| private IgniteBinary binaries; |
| |
| /** Locally cached metadata. This local cache is managed by exchanging discovery custom events. */ |
| private final ConcurrentMap<Integer, BinaryMetadataHolder> metadataLocCache = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private BinaryMetadataTransport transport; |
| |
| /** Cached affinity key field names. */ |
| private final ConcurrentHashMap<Integer, T1<BinaryField>> affKeyFields = new ConcurrentHashMap<>(); |
| |
| /* |
| * Static initializer |
| */ |
| static { |
| IMMUTABLE_CLS.add(String.class); |
| IMMUTABLE_CLS.add(Boolean.class); |
| IMMUTABLE_CLS.add(Byte.class); |
| IMMUTABLE_CLS.add(Short.class); |
| IMMUTABLE_CLS.add(Character.class); |
| IMMUTABLE_CLS.add(Integer.class); |
| IMMUTABLE_CLS.add(Long.class); |
| IMMUTABLE_CLS.add(Float.class); |
| IMMUTABLE_CLS.add(Double.class); |
| IMMUTABLE_CLS.add(UUID.class); |
| IMMUTABLE_CLS.add(IgniteUuid.class); |
| IMMUTABLE_CLS.add(BigDecimal.class); |
| } |
| |
| /** |
| * @param ctx Kernal context. |
| */ |
| @SuppressWarnings("deprecation") |
| public CacheObjectBinaryProcessorImpl(GridKernalContext ctx) { |
| super(ctx); |
| |
| marsh = ctx.grid().configuration().getMarshaller(); |
| } |
| |
| /** |
| * @param igniteWorkDir Basic ignite working directory. |
| * @param consId Node consistent id. |
| * @return Working directory. |
| */ |
| public static File resolveBinaryWorkDir(String igniteWorkDir, String consId) { |
| File workDir = binaryWorkDir(igniteWorkDir, consId); |
| |
| if (!U.mkdirs(workDir)) |
| throw new IgniteException("Could not create directory for binary metadata: " + workDir); |
| |
| return workDir; |
| } |
| |
| /** |
| * @param igniteWorkDir Basic ignite working directory. |
| * @param consId Node consistent id. |
| * @return Working directory. |
| */ |
| public static File binaryWorkDir(String igniteWorkDir, String consId) { |
| if (F.isEmpty(igniteWorkDir) || F.isEmpty(consId)) { |
| throw new IgniteException("Work directory or consistent id has not been set " + |
| "[igniteWorkDir=" + igniteWorkDir + ", consId=" + consId + ']'); |
| } |
| |
| return Paths.get(igniteWorkDir, DataStorageConfiguration.DFLT_BINARY_METADATA_PATH, consId).toFile(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void start() throws IgniteCheckedException { |
| if (marsh instanceof BinaryMarshaller) { |
| if (!ctx.clientNode()) { |
| metadataFileStore = new BinaryMetadataFileStore(metadataLocCache, |
| ctx, |
| log, |
| CU.isPersistenceEnabled(ctx.config()) && binaryMetadataFileStoreDir == null ? |
| resolveBinaryWorkDir(ctx.config().getWorkDirectory(), |
| ctx.pdsFolderResolver().resolveFolders().folderName()) : |
| binaryMetadataFileStoreDir); |
| |
| metadataFileStore.start(); |
| } |
| |
| BinaryMetadataHandler metaHnd = new BinaryMetadataHandler() { |
| @Override public void addMeta( |
| int typeId, |
| BinaryType newMeta, |
| boolean failIfUnregistered) throws BinaryObjectException { |
| assert newMeta != null; |
| assert newMeta instanceof BinaryTypeImpl; |
| |
| if (!discoveryStarted) { |
| BinaryMetadataHolder holder = metadataLocCache.get(typeId); |
| |
| BinaryMetadata oldMeta = holder != null ? holder.metadata() : null; |
| |
| BinaryMetadata mergedMeta = mergeMetadata(oldMeta, ((BinaryTypeImpl)newMeta).metadata()); |
| |
| if (oldMeta != mergedMeta) |
| metadataLocCache.put(typeId, new BinaryMetadataHolder(mergedMeta, 0, 0)); |
| |
| return; |
| } |
| |
| BinaryMetadata newMeta0 = ((BinaryTypeImpl)newMeta).metadata(); |
| |
| CacheObjectBinaryProcessorImpl.this.addMeta( |
| typeId, |
| newMeta0.wrap(binaryCtx), |
| failIfUnregistered |
| ); |
| } |
| |
| @Override public void addMetaLocally(int typeId, BinaryType meta, boolean failIfUnregistered) |
| throws BinaryObjectException { |
| CacheObjectBinaryProcessorImpl.this.addMetaLocally(typeId, meta); |
| } |
| |
| @Override public BinaryType metadata(int typeId) throws BinaryObjectException { |
| return CacheObjectBinaryProcessorImpl.this.metadata(typeId); |
| } |
| |
| @Override public BinaryMetadata metadata0(int typeId) throws BinaryObjectException { |
| return CacheObjectBinaryProcessorImpl.this.metadata0(typeId); |
| } |
| |
| @Override public BinaryType metadata(int typeId, int schemaId) throws BinaryObjectException { |
| return CacheObjectBinaryProcessorImpl.this.metadata(typeId, schemaId); |
| } |
| |
| @Override public Collection<BinaryType> metadata() throws BinaryObjectException { |
| return CacheObjectBinaryProcessorImpl.this.metadata(); |
| } |
| }; |
| |
| BinaryMarshaller bMarsh0 = (BinaryMarshaller)marsh; |
| |
| binaryCtx = useTestBinaryCtx ? |
| new TestBinaryContext(metaHnd, ctx.config(), ctx.log(BinaryContext.class)) : |
| new BinaryContext(metaHnd, ctx.config(), ctx.log(BinaryContext.class)); |
| |
| transport = new BinaryMetadataTransport(metadataLocCache, metadataFileStore, binaryCtx, ctx, log); |
| |
| IgniteUtils.invoke(BinaryMarshaller.class, bMarsh0, "setBinaryContext", binaryCtx, ctx.config()); |
| |
| binaryMarsh = new GridBinaryMarshaller(binaryCtx); |
| |
| binaries = new IgniteBinaryImpl(ctx, this); |
| |
| if (!getBoolean(IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK)) { |
| BinaryConfiguration bCfg = ctx.config().getBinaryConfiguration(); |
| |
| if (bCfg != null) { |
| Map<String, Object> map = new HashMap<>(); |
| |
| map.put("globIdMapper", bCfg.getIdMapper() != null ? bCfg.getIdMapper().getClass().getName() : null); |
| map.put("globSerializer", bCfg.getSerializer() != null ? bCfg.getSerializer().getClass() : null); |
| map.put("compactFooter", bCfg.isCompactFooter()); |
| |
| if (bCfg.getTypeConfigurations() != null) { |
| Map<Object, Object> typeCfgsMap = new HashMap<>(); |
| |
| for (BinaryTypeConfiguration c : bCfg.getTypeConfigurations()) { |
| typeCfgsMap.put( |
| c.getTypeName() != null, |
| Arrays.asList( |
| c.getIdMapper() != null ? c.getIdMapper().getClass() : null, |
| c.getSerializer() != null ? c.getSerializer().getClass() : null, |
| c.isEnum() |
| ) |
| ); |
| |
| if (c.isEnum()) |
| BinaryUtils.validateEnumValues(c.getTypeName(), c.getEnumValues()); |
| } |
| |
| map.put("typeCfgs", typeCfgsMap); |
| } |
| |
| ctx.addNodeAttribute(IgniteNodeAttributes.ATTR_BINARY_CONFIGURATION, map); |
| } |
| } |
| |
| if (!ctx.clientNode()) |
| metadataFileStore.restoreMetadata(); |
| } |
| } |
| |
| /** |
| * @param lsnr Listener. |
| */ |
| public void addBinaryMetadataUpdateListener(BinaryMetadataUpdatedListener lsnr) { |
| if (transport != null) |
| transport.addBinaryMetadataUpdateListener(lsnr); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void stop(boolean cancel) { |
| if (transport != null) |
| transport.stop(); |
| |
| if (metadataFileStore != null) |
| metadataFileStore.stop(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onDisconnected(IgniteFuture<?> reconnectFut) { |
| this.reconnectFut = reconnectFut; |
| |
| if (transport != null) |
| transport.onDisconnected(); |
| |
| binaryContext().unregisterUserTypeDescriptors(); |
| binaryContext().unregisterBinarySchemas(); |
| |
| metadataLocCache.clear(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<?> onReconnected(boolean clusterRestarted) throws IgniteCheckedException { |
| reconnectFut = null; |
| |
| return super.onReconnected(clusterRestarted); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onKernalStart(boolean active) throws IgniteCheckedException { |
| super.onKernalStart(active); |
| |
| discoveryStarted = true; |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public CacheObject prepareForCache(@Nullable CacheObject obj, GridCacheContext cctx) { |
| if (obj == null) |
| return null; |
| |
| return obj.prepareForCache(cctx.cacheObjectContext()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int typeId(String typeName) { |
| if (binaryCtx == null) |
| return 0; |
| |
| return binaryCtx.typeId(typeName); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean immutable(Object obj) { |
| assert obj != null; |
| |
| return IMMUTABLE_CLS.contains(obj.getClass()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onContinuousProcessorStarted(GridKernalContext ctx) { |
| // No-op. |
| } |
| |
| /** |
| * @param obj Object. |
| * @return Bytes. |
| * @throws BinaryObjectException If failed. |
| */ |
| public byte[] marshal(@Nullable Object obj) throws BinaryObjectException { |
| byte[] arr = binaryMarsh.marshal(obj, false); |
| |
| assert arr.length > 0; |
| |
| return arr; |
| } |
| |
| /** |
| * @param ptr Off-heap pointer. |
| * @param forceHeap If {@code true} creates heap-based object. |
| * @return Object. |
| * @throws BinaryObjectException If failed. |
| */ |
| public Object unmarshal(long ptr, boolean forceHeap) throws BinaryObjectException { |
| assert ptr > 0 : ptr; |
| |
| int size = GridUnsafe.getInt(ptr); |
| |
| ptr += 4; |
| |
| byte type = GridUnsafe.getByte(ptr++); |
| |
| if (type != CacheObject.TYPE_BYTE_ARR) { |
| assert size > 0 : size; |
| |
| BinaryInputStream in = new BinaryOffheapInputStream(ptr, size, forceHeap); |
| |
| return binaryMarsh.unmarshal(in); |
| } |
| else |
| return U.copyMemory(ptr, size); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object marshalToBinary( |
| @Nullable Object obj, |
| boolean failIfUnregistered |
| ) throws BinaryObjectException { |
| if (obj == null) |
| return null; |
| |
| if (BinaryUtils.isBinaryType(obj.getClass())) |
| return obj; |
| |
| if (obj instanceof Object[]) { |
| Object[] arr = (Object[])obj; |
| |
| Object[] pArr = new Object[arr.length]; |
| |
| for (int i = 0; i < arr.length; i++) |
| pArr[i] = marshalToBinary(arr[i], failIfUnregistered); |
| |
| return pArr; |
| } |
| |
| if (obj instanceof IgniteBiTuple) { |
| IgniteBiTuple tup = (IgniteBiTuple)obj; |
| |
| if (obj instanceof T2) |
| return new T2<>(marshalToBinary(tup.get1(), failIfUnregistered), |
| marshalToBinary(tup.get2(), failIfUnregistered)); |
| |
| return new IgniteBiTuple<>(marshalToBinary(tup.get1(), failIfUnregistered), |
| marshalToBinary(tup.get2(), failIfUnregistered)); |
| } |
| |
| { |
| Collection<Object> pCol = BinaryUtils.newKnownCollection(obj); |
| |
| if (pCol != null) { |
| Collection<?> col = (Collection<?>)obj; |
| |
| for (Object item : col) |
| pCol.add(marshalToBinary(item, failIfUnregistered)); |
| |
| return (pCol instanceof MutableSingletonList) ? U.convertToSingletonList(pCol) : pCol; |
| } |
| } |
| |
| { |
| Map<Object, Object> pMap = BinaryUtils.newKnownMap(obj); |
| |
| if (pMap != null) { |
| Map<?, ?> map = (Map<?, ?>)obj; |
| |
| for (Map.Entry<?, ?> e : map.entrySet()) |
| pMap.put(marshalToBinary(e.getKey(), failIfUnregistered), |
| marshalToBinary(e.getValue(), failIfUnregistered)); |
| |
| return pMap; |
| } |
| } |
| |
| if (obj instanceof Map.Entry) { |
| Map.Entry<?, ?> e = (Map.Entry<?, ?>)obj; |
| |
| return new GridMapEntry<>(marshalToBinary(e.getKey(), failIfUnregistered), |
| marshalToBinary(e.getValue(), failIfUnregistered)); |
| } |
| |
| if (binaryMarsh.mustDeserialize(obj)) |
| return obj; // No need to go through marshal-unmarshal because result will be the same as initial object. |
| |
| byte[] arr = binaryMarsh.marshal(obj, failIfUnregistered); |
| |
| assert arr.length > 0; |
| |
| Object obj0 = binaryMarsh.unmarshal(arr, null); |
| |
| // Possible if a class has writeObject method. |
| if (obj0 instanceof BinaryObjectImpl) |
| ((BinaryObjectImpl)obj0).detachAllowed(true); |
| |
| return obj0; |
| } |
| |
| /** |
| * @return Marshaller. |
| */ |
| public GridBinaryMarshaller marshaller() { |
| return binaryMarsh; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public BinaryObjectBuilder builder(String clsName) { |
| return new BinaryObjectBuilderImpl(binaryCtx, clsName); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public BinaryObjectBuilder builder(BinaryObject binaryObj) { |
| return BinaryObjectBuilderImpl.wrap(binaryObj); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void updateMetadata(int typeId, String typeName, @Nullable String affKeyFieldName, |
| Map<String, BinaryFieldMetadata> fieldTypeIds, boolean isEnum, @Nullable Map<String, Integer> enumMap) |
| throws BinaryObjectException { |
| BinaryMetadata meta = new BinaryMetadata(typeId, typeName, fieldTypeIds, affKeyFieldName, null, isEnum, |
| enumMap); |
| |
| binaryCtx.updateMetadata(typeId, meta, false); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void addMeta(final int typeId, final BinaryType newMeta, boolean failIfUnregistered) |
| throws BinaryObjectException { |
| assert newMeta != null; |
| assert newMeta instanceof BinaryTypeImpl; |
| |
| BinaryMetadata newMeta0 = ((BinaryTypeImpl)newMeta).metadata(); |
| |
| if (failIfUnregistered) { |
| failIfUnregistered(typeId, newMeta0); |
| |
| return; |
| } |
| |
| try { |
| GridFutureAdapter<MetadataUpdateResult> fut = transport.requestMetadataUpdate(newMeta0); |
| |
| if (fut == null) { |
| if (log.isDebugEnabled()) { |
| log.debug("Metadata update was skipped [typeId=" + typeId |
| + ", typeName=" + newMeta.typeName() + ']'); |
| } |
| |
| return; |
| } |
| |
| long t0 = System.nanoTime(); |
| |
| MetadataUpdateResult res = fut.get(); |
| |
| if (log.isDebugEnabled()) { |
| IgniteInternalTx tx = ctx.cache().context().tm().tx(); |
| |
| log.debug("Completed metadata update [typeId=" + typeId + |
| ", typeName=" + newMeta.typeName() + |
| ", waitTime=" + MILLISECONDS.convert(System.nanoTime() - t0, NANOSECONDS) + "ms" + |
| ", fut=" + fut + |
| ", tx=" + CU.txString(tx) + |
| ']'); |
| } |
| |
| assert res != null; |
| |
| if (res.rejected()) |
| throw res.error(); |
| else if (!ctx.clientNode()) |
| metadataFileStore.waitForWriteCompletion(typeId, res.typeVersion()); |
| } |
| catch (IgniteCheckedException e) { |
| IgniteCheckedException ex = e; |
| |
| if (ctx.isStopping()) { |
| ex = new NodeStoppingException("Node is stopping."); |
| |
| ex.addSuppressed(e); |
| } |
| |
| throw new BinaryObjectException("Failed to update metadata for type: " + newMeta.typeName(), ex); |
| } |
| } |
| |
| /** |
| * Throw specific exception if given binary metadata is unregistered. |
| * |
| * @param typeId Type id. |
| * @param newMeta0 Expected binary metadata. |
| */ |
| private void failIfUnregistered(int typeId, BinaryMetadata newMeta0) { |
| BinaryMetadataHolder metaHolder = metadataLocCache.get(typeId); |
| |
| BinaryMetadata oldMeta = metaHolder != null ? metaHolder.metadata() : null; |
| |
| BinaryMetadata mergedMeta = mergeMetadata(oldMeta, newMeta0); |
| |
| if (mergedMeta != oldMeta) |
| throw new UnregisteredBinaryTypeException(typeId, mergedMeta); |
| |
| if (metaHolder.pendingVersion() == metaHolder.acceptedVersion()) |
| return; |
| |
| // Metadata locally is up-to-date. Waiting for updating metadata in an entire cluster, if necessary. |
| GridFutureAdapter<MetadataUpdateResult> fut = transport.awaitMetadataUpdate(typeId, metaHolder.pendingVersion()); |
| |
| if (!fut.isDone()) |
| throw new UnregisteredBinaryTypeException(typeId, fut); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void addMetaLocally(int typeId, BinaryType newMeta) throws BinaryObjectException { |
| assert newMeta != null; |
| assert newMeta instanceof BinaryTypeImpl; |
| |
| BinaryMetadata newMeta0 = ((BinaryTypeImpl)newMeta).metadata(); |
| |
| BinaryMetadataHolder metaHolder = metadataLocCache.get(typeId); |
| |
| BinaryMetadata oldMeta = metaHolder != null ? metaHolder.metadata() : null; |
| |
| try { |
| BinaryMetadata mergedMeta = mergeMetadata(oldMeta, newMeta0); |
| |
| if (!ctx.clientNode()) |
| metadataFileStore.mergeAndWriteMetadata(mergedMeta); |
| |
| metadataLocCache.put(typeId, new BinaryMetadataHolder(mergedMeta, 0, 0)); |
| } |
| catch (BinaryObjectException e) { |
| throw new BinaryObjectException("New binary metadata is incompatible with binary metadata" + |
| " persisted locally." + |
| " Consider cleaning up persisted metadata from <workDir>/db/binary_meta directory.", e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public BinaryType metadata(final int typeId) { |
| BinaryMetadata meta = metadata0(typeId); |
| |
| return meta != null ? meta.wrap(binaryCtx) : null; |
| } |
| |
| /** |
| * Forces caller thread to wait for binary metadata write operation for given type ID. |
| * |
| * In case of in-memory mode this method becomes a No-op as no binary metadata is written to disk in this mode. |
| * |
| * @param typeId ID of binary type to wait for metadata write operation. |
| */ |
| public void waitMetadataWriteIfNeeded(final int typeId) { |
| if (metadataFileStore == null) |
| return; |
| |
| BinaryMetadataHolder hldr = metadataLocCache.get(typeId); |
| |
| if (hldr != null) { |
| try { |
| metadataFileStore.waitForWriteCompletion(typeId, hldr.pendingVersion()); |
| } |
| catch (IgniteCheckedException e) { |
| log.warning("Failed to wait for metadata write operation for [typeId=" + typeId + |
| ", typeVer=" + hldr.acceptedVersion() + ']', e); |
| } |
| } |
| } |
| |
| /** |
| * @param typeId Type ID. |
| * @return Metadata. |
| * @throws IgniteException In case of error. |
| */ |
| @Nullable public BinaryMetadata metadata0(final int typeId) { |
| BinaryMetadataHolder holder = metadataLocCache.get(typeId); |
| |
| IgniteThread curThread = IgniteThread.current(); |
| |
| if (holder == null && (curThread == null || !curThread.isForbiddenToRequestBinaryMetadata())) { |
| if (ctx.clientNode()) { |
| try { |
| transport.requestUpToDateMetadata(typeId).get(); |
| |
| holder = metadataLocCache.get(typeId); |
| } |
| catch (IgniteCheckedException ignored) { |
| // No-op. |
| } |
| } |
| } |
| |
| if (holder != null) { |
| if (holder.removing()) { |
| GridFutureAdapter<MetadataUpdateResult> fut = transport.awaitMetadataRemove(typeId); |
| |
| try { |
| fut.get(); |
| } |
| catch (IgniteCheckedException ignored) { |
| // No-op. |
| } |
| |
| return null; |
| } |
| |
| if (curThread instanceof IgniteDiscoveryThread || (curThread != null && curThread.isForbiddenToRequestBinaryMetadata())) |
| return holder.metadata(); |
| |
| if (holder.pendingVersion() - holder.acceptedVersion() > 0) { |
| GridFutureAdapter<MetadataUpdateResult> fut = transport.awaitMetadataUpdate(typeId, holder.pendingVersion()); |
| |
| if (log.isDebugEnabled() && !fut.isDone()) |
| log.debug("Waiting for update for" + |
| " [typeId=" + typeId + |
| ", pendingVer=" + holder.pendingVersion() + |
| ", acceptedVer=" + holder.acceptedVersion() + "]"); |
| |
| try { |
| fut.get(); |
| } |
| catch (IgniteCheckedException ignored) { |
| // No-op. |
| } |
| } |
| else if (metadataFileStore != null) { |
| try { |
| metadataFileStore.waitForWriteCompletion(typeId, holder.pendingVersion()); |
| } |
| catch (IgniteCheckedException e) { |
| log.warning("Failed to wait for metadata write operation for [typeId=" + typeId + |
| ", typeVer=" + holder.acceptedVersion() + ']', e); |
| |
| return null; |
| } |
| } |
| |
| return holder.metadata(); |
| } |
| else |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public BinaryType metadata(final int typeId, final int schemaId) { |
| BinaryMetadataHolder holder = metadataLocCache.get(typeId); |
| |
| if (ctx.clientNode()) { |
| if (holder == null || !holder.metadata().hasSchema(schemaId)) { |
| if (log.isDebugEnabled()) |
| log.debug("Waiting for client metadata update" + |
| " [typeId=" + typeId |
| + ", schemaId=" + schemaId |
| + ", pendingVer=" + (holder == null ? "NA" : holder.pendingVersion()) |
| + ", acceptedVer=" + (holder == null ? "NA" : holder.acceptedVersion()) + ']'); |
| |
| try { |
| transport.requestUpToDateMetadata(typeId).get(); |
| } |
| catch (IgniteCheckedException ignored) { |
| // No-op. |
| } |
| |
| holder = metadataLocCache.get(typeId); |
| |
| IgniteFuture<?> reconnectFut0 = reconnectFut; |
| |
| if (holder == null && reconnectFut0 != null) |
| throw new IgniteClientDisconnectedException(reconnectFut0, "Client node disconnected."); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Finished waiting for client metadata update" + |
| " [typeId=" + typeId |
| + ", schemaId=" + schemaId |
| + ", pendingVer=" + (holder == null ? "NA" : holder.pendingVersion()) |
| + ", acceptedVer=" + (holder == null ? "NA" : holder.acceptedVersion()) + ']'); |
| } |
| } |
| else { |
| if (holder != null && IgniteThread.current() instanceof IgniteDiscoveryThread) |
| return holder.metadata().wrap(binaryCtx); |
| else if (holder != null && (holder.pendingVersion() - holder.acceptedVersion() > 0)) { |
| if (log.isDebugEnabled()) |
| log.debug("Waiting for metadata update" + |
| " [typeId=" + typeId |
| + ", schemaId=" + schemaId |
| + ", pendingVer=" + holder.pendingVersion() |
| + ", acceptedVer=" + holder.acceptedVersion() + ']'); |
| |
| long t0 = System.nanoTime(); |
| |
| GridFutureAdapter<MetadataUpdateResult> fut = transport.awaitMetadataUpdate( |
| typeId, |
| holder.pendingVersion()); |
| |
| try { |
| fut.get(); |
| } |
| catch (IgniteCheckedException e) { |
| log.error("Failed to wait for metadata update [typeId=" + typeId + ", schemaId=" + schemaId + ']', e); |
| } |
| |
| if (log.isDebugEnabled()) |
| log.debug("Finished waiting for metadata update" + |
| " [typeId=" + typeId |
| + ", waitTime=" + NANOSECONDS.convert(System.nanoTime() - t0, MILLISECONDS) + "ms" |
| + ", schemaId=" + schemaId |
| + ", pendingVer=" + holder.pendingVersion() |
| + ", acceptedVer=" + holder.acceptedVersion() + ']'); |
| |
| holder = metadataLocCache.get(typeId); |
| } |
| else if (holder == null || !holder.metadata().hasSchema(schemaId)) { |
| // Last resort waiting. |
| U.warn(log, |
| "Schema is missing while no metadata updates are in progress " + |
| "(will wait for schema update within timeout defined by " + IGNITE_WAIT_SCHEMA_UPDATE + " system property)" + |
| " [typeId=" + typeId |
| + ", missingSchemaId=" + schemaId |
| + ", pendingVer=" + (holder == null ? "NA" : holder.pendingVersion()) |
| + ", acceptedVer=" + (holder == null ? "NA" : holder.acceptedVersion()) |
| + ", binMetaUpdateTimeout=" + waitSchemaTimeout + ']'); |
| |
| long t0 = System.nanoTime(); |
| |
| GridFutureAdapter<?> fut = transport.awaitSchemaUpdate(typeId, schemaId); |
| |
| try { |
| fut.get(waitSchemaTimeout); |
| } |
| catch (IgniteFutureTimeoutCheckedException e) { |
| log.error("Timed out while waiting for schema update [typeId=" + typeId + ", schemaId=" + |
| schemaId + ']'); |
| } |
| catch (IgniteCheckedException ignored) { |
| // No-op. |
| } |
| |
| holder = metadataLocCache.get(typeId); |
| |
| if (log.isDebugEnabled() && holder != null && holder.metadata().hasSchema(schemaId)) |
| log.debug("Found the schema after wait" + |
| " [typeId=" + typeId |
| + ", waitTime=" + NANOSECONDS.convert(System.nanoTime() - t0, MILLISECONDS) + "ms" |
| + ", schemaId=" + schemaId |
| + ", pendingVer=" + holder.pendingVersion() |
| + ", acceptedVer=" + holder.acceptedVersion() + ']'); |
| } |
| } |
| |
| if (holder != null && metadataFileStore != null) { |
| try { |
| metadataFileStore.waitForWriteCompletion(typeId, holder.pendingVersion()); |
| } |
| catch (IgniteCheckedException e) { |
| log.warning("Failed to wait for metadata write operation for [typeId=" + typeId + |
| ", typeVer=" + holder.acceptedVersion() + ']', e); |
| |
| return null; |
| } |
| } |
| |
| return holder != null ? holder.metadata().wrap(binaryCtx) : null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Map<Integer, BinaryType> metadata(Collection<Integer> typeIds) |
| throws BinaryObjectException { |
| try { |
| Map<Integer, BinaryType> res = U.newHashMap(metadataLocCache.size()); |
| |
| for (Map.Entry<Integer, BinaryMetadataHolder> e : metadataLocCache.entrySet()) |
| res.put(e.getKey(), e.getValue().metadata().wrap(binaryCtx)); |
| |
| return res; |
| } |
| catch (CacheException e) { |
| throw new BinaryObjectException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Collection<BinaryType> metadata() throws BinaryObjectException { |
| return F.viewReadOnly(metadataLocCache.values(), new IgniteClosure<BinaryMetadataHolder, BinaryType>() { |
| @Override public BinaryType apply(BinaryMetadataHolder metaHolder) { |
| return metaHolder.metadata().wrap(binaryCtx); |
| } |
| }); |
| } |
| |
| /** |
| * @return Cluster binary metadata. |
| * @throws BinaryObjectException on error. |
| */ |
| public Collection<BinaryMetadata> binaryMetadata() throws BinaryObjectException { |
| return F.viewReadOnly(metadataLocCache.values(), new IgniteClosure<BinaryMetadataHolder, BinaryMetadata>() { |
| @Override public BinaryMetadata apply(BinaryMetadataHolder metaHolder) { |
| return metaHolder.metadata(); |
| } |
| }); |
| } |
| |
| /** |
| * @return Binary metadata for specified type. |
| * @throws BinaryObjectException on error. |
| */ |
| public BinaryMetadata binaryMetadata(int typeId) throws BinaryObjectException { |
| BinaryMetadataHolder hld = metadataLocCache.get(typeId); |
| |
| return hld != null ? hld.metadata() : null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void saveMetadata(Collection<BinaryType> types, File dir) { |
| try { |
| BinaryMetadataFileStore writer = new BinaryMetadataFileStore(new ConcurrentHashMap<>(), |
| ctx, |
| log, |
| resolveBinaryWorkDir(dir.getAbsolutePath(), |
| ctx.pdsFolderResolver().resolveFolders().folderName())); |
| |
| for (BinaryType type : types) |
| writer.mergeAndWriteMetadata(((BinaryTypeImpl)type).metadata()); |
| } |
| catch (IgniteCheckedException e) { |
| throw new IgniteException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public BinaryObject buildEnum(String typeName, int ord) throws BinaryObjectException { |
| A.notNullOrEmpty(typeName, "enum type name"); |
| |
| int typeId = binaryCtx.typeId(typeName); |
| |
| typeName = binaryCtx.userTypeName(typeName); |
| |
| updateMetadata(typeId, typeName, null, null, true, null); |
| |
| return new BinaryEnumObjectImpl(binaryCtx, typeId, null, ord); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public BinaryObject buildEnum(String typeName, String name) throws BinaryObjectException { |
| A.notNullOrEmpty(typeName, "enum type name"); |
| A.notNullOrEmpty(name, "enum name"); |
| |
| int typeId = binaryCtx.typeId(typeName); |
| |
| BinaryMetadata metadata = metadata0(typeId); |
| |
| if (metadata == null) |
| throw new BinaryObjectException("Failed to get metadata for type [typeId=" + |
| typeId + ", typeName='" + typeName + "']"); |
| |
| Integer ordinal = metadata.getEnumOrdinalByName(name); |
| |
| typeName = binaryCtx.userTypeName(typeName); |
| |
| if (ordinal == null) |
| throw new BinaryObjectException("Failed to resolve enum ordinal by name [typeId=" + |
| typeId + ", typeName='" + typeName + "', name='" + name + "']"); |
| |
| return new BinaryEnumObjectImpl(binaryCtx, typeId, null, ordinal); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public BinaryType registerEnum(String typeName, Map<String, Integer> vals) throws BinaryObjectException { |
| A.notNullOrEmpty(typeName, "enum type name"); |
| |
| int typeId = binaryCtx.typeId(typeName); |
| |
| typeName = binaryCtx.userTypeName(typeName); |
| |
| BinaryUtils.validateEnumValues(typeName, vals); |
| |
| updateMetadata(typeId, typeName, null, null, true, vals); |
| |
| return binaryCtx.metadata(typeId); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteBinary binary() throws IgniteException { |
| return binaries; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean isBinaryObject(Object obj) { |
| return obj instanceof BinaryObject; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean isBinaryEnabled(CacheConfiguration<?, ?> ccfg) { |
| return marsh instanceof BinaryMarshaller; |
| } |
| |
| /** |
| * Get affinity key field. |
| * |
| * @param typeId Binary object type ID. |
| * @return Affinity key. |
| */ |
| public BinaryField affinityKeyField(int typeId) { |
| // Fast path for already cached field. |
| T1<BinaryField> fieldHolder = affKeyFields.get(typeId); |
| |
| if (fieldHolder != null) |
| return fieldHolder.get(); |
| |
| // Slow path if affinity field is not cached yet. |
| String name = binaryCtx.affinityKeyFieldName(typeId); |
| |
| if (name != null) { |
| BinaryField field = binaryCtx.createField(typeId, name); |
| |
| affKeyFields.putIfAbsent(typeId, new T1<>(field)); |
| |
| return field; |
| } |
| else { |
| affKeyFields.putIfAbsent(typeId, new T1<>(null)); |
| |
| return null; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int typeId(Object obj) { |
| if (obj == null) |
| return 0; |
| |
| return isBinaryObject(obj) ? ((BinaryObjectEx)obj).typeId() : typeId(obj.getClass().getSimpleName()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object field(Object obj, String fieldName) { |
| if (obj == null) |
| return null; |
| |
| return isBinaryObject(obj) ? ((BinaryObject)obj).field(fieldName) : null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean hasField(Object obj, String fieldName) { |
| return obj != null && ((BinaryObject)obj).hasField(fieldName); |
| } |
| |
| /** |
| * @return Binary context. |
| */ |
| public BinaryContext binaryContext() { |
| return binaryCtx; |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("deprecation") |
| @Override public CacheObjectContext contextForCache(CacheConfiguration ccfg) throws IgniteCheckedException { |
| assert ccfg != null; |
| |
| boolean storeVal = !ccfg.isCopyOnRead() || (!isBinaryEnabled(ccfg) && |
| (QueryUtils.isEnabled(ccfg) || ctx.config().isPeerClassLoadingEnabled())); |
| |
| boolean binaryEnabled = marsh instanceof BinaryMarshaller && !GridCacheUtils.isSystemCache(ccfg.getName()); |
| |
| AffinityKeyMapper cacheAffMapper = ccfg.getAffinityMapper(); |
| |
| AffinityKeyMapper dfltAffMapper = binaryEnabled ? |
| new CacheDefaultBinaryAffinityKeyMapper(ccfg.getKeyConfiguration()) : |
| new GridCacheDefaultAffinityKeyMapper(); |
| |
| ctx.resource().injectGeneric(dfltAffMapper); |
| |
| return new CacheObjectContext(ctx, |
| ccfg.getName(), |
| dfltAffMapper, |
| QueryUtils.isCustomAffinityMapper(ccfg.getAffinityMapper()), |
| ccfg.isCopyOnRead(), |
| storeVal, |
| ctx.config().isPeerClassLoadingEnabled() && !isBinaryEnabled(ccfg), |
| binaryEnabled |
| ); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public byte[] marshal(CacheObjectValueContext ctx, Object val) throws IgniteCheckedException { |
| if (!ctx.binaryEnabled() || binaryMarsh == null) |
| return CU.marshal(ctx.kernalContext().cache().context(), ctx.addDeploymentInfo(), val); |
| |
| byte[] arr = binaryMarsh.marshal(val, false); |
| |
| assert arr.length > 0; |
| |
| return arr; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object unmarshal(CacheObjectValueContext ctx, byte[] bytes, ClassLoader clsLdr) |
| throws IgniteCheckedException { |
| if (!ctx.binaryEnabled() || binaryMarsh == null) |
| return U.unmarshal(ctx.kernalContext(), bytes, U.resolveClassLoader(clsLdr, ctx.kernalContext().config())); |
| |
| return binaryMarsh.unmarshal(bytes, clsLdr); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public KeyCacheObject toCacheKeyObject(CacheObjectContext ctx, @Nullable GridCacheContext cctx, |
| Object obj, boolean userObj) { |
| if (!ctx.binaryEnabled()) { |
| if (obj instanceof KeyCacheObject) { |
| KeyCacheObject key = (KeyCacheObject)obj; |
| |
| if (key.partition() == -1) |
| // Assume all KeyCacheObjects except BinaryObject can not be reused for another cache. |
| key.partition(partition(ctx, cctx, key)); |
| |
| return (KeyCacheObject)obj; |
| } |
| |
| return toCacheKeyObject0(ctx, cctx, obj, userObj); |
| } |
| |
| if (obj instanceof KeyCacheObject) { |
| KeyCacheObject key = (KeyCacheObject)obj; |
| |
| if (key instanceof BinaryObjectImpl) { |
| // Need to create a copy because the key can be reused at the application layer after that (IGNITE-3505). |
| key = key.copy(partition(ctx, cctx, key)); |
| } |
| else if (key.partition() == -1) |
| // Assume others KeyCacheObjects can not be reused for another cache. |
| key.partition(partition(ctx, cctx, key)); |
| |
| return key; |
| } |
| |
| obj = toBinary(obj, false); |
| |
| if (obj instanceof BinaryObjectImpl) { |
| ((KeyCacheObject) obj).partition(partition(ctx, cctx, obj)); |
| |
| return (KeyCacheObject)obj; |
| } |
| |
| return toCacheKeyObject0(ctx, cctx, obj, userObj); |
| } |
| |
| /** |
| * @param obj Object. |
| * @param userObj If {@code true} then given object is object provided by user and should be copied |
| * before stored in cache. |
| * @return Key cache object. |
| */ |
| protected KeyCacheObject toCacheKeyObject0(CacheObjectContext ctx, |
| @Nullable GridCacheContext cctx, |
| Object obj, |
| boolean userObj) { |
| int part = partition(ctx, cctx, obj); |
| |
| if (!userObj) |
| return new KeyCacheObjectImpl(obj, null, part); |
| |
| return new UserKeyCacheObjectImpl(obj, part); |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public CacheObject toCacheObject(CacheObjectContext ctx, @Nullable Object obj, |
| boolean userObj, boolean failIfUnregistered) { |
| if (!ctx.binaryEnabled()) { |
| if (obj == null || obj instanceof CacheObject) |
| return (CacheObject)obj; |
| |
| return toCacheObject0(obj, userObj); |
| } |
| |
| if (obj == null || obj instanceof CacheObject) |
| return (CacheObject)obj; |
| |
| obj = toBinary(obj, failIfUnregistered); |
| |
| if (obj instanceof CacheObject) |
| return (CacheObject)obj; |
| |
| return toCacheObject0(obj, userObj); |
| } |
| |
| /** |
| * @param obj Object. |
| * @param userObj If {@code true} then given object is object provided by user and should be copied |
| * before stored in cache. |
| * @return Cache object. |
| */ |
| private CacheObject toCacheObject0(@Nullable Object obj, boolean userObj) { |
| assert obj != null; |
| |
| if (obj instanceof byte[]) { |
| if (!userObj) |
| return new CacheObjectByteArrayImpl((byte[])obj); |
| |
| return new UserCacheObjectByteArrayImpl((byte[])obj); |
| } |
| |
| if (!userObj) |
| return new CacheObjectImpl(obj, null); |
| |
| return new UserCacheObjectImpl(obj, null); |
| } |
| |
| /** |
| * @param ctx Cache objects context. |
| * @param cctx Cache context. |
| * @param obj Object. |
| * @return Object partition. |
| */ |
| private int partition(CacheObjectContext ctx, @Nullable GridCacheContext cctx, Object obj) { |
| try { |
| return cctx != null ? |
| cctx.affinity().partition(obj, false) : |
| ctx.kernalContext().affinity().partition0(ctx.cacheName(), obj, null); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to get partition", e); |
| |
| return -1; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public CacheObject toCacheObject(CacheObjectContext ctx, byte type, byte[] bytes) { |
| switch (type) { |
| case BinaryObjectImpl.TYPE_BINARY: |
| return new BinaryObjectImpl(binaryContext(), bytes, 0); |
| |
| case BinaryObjectImpl.TYPE_BINARY_ENUM: |
| return new BinaryEnumObjectImpl(binaryContext(), bytes); |
| |
| case CacheObject.TYPE_BYTE_ARR: |
| return new CacheObjectByteArrayImpl(bytes); |
| |
| case CacheObject.TYPE_REGULAR: |
| return new CacheObjectImpl(null, bytes); |
| } |
| |
| throw new IllegalArgumentException("Invalid object type: " + type); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public KeyCacheObject toKeyCacheObject(CacheObjectContext ctx, byte type, byte[] bytes) |
| throws IgniteCheckedException { |
| switch (type) { |
| case BinaryObjectImpl.TYPE_BINARY: |
| return new BinaryObjectImpl(binaryContext(), bytes, 0); |
| |
| case CacheObject.TYPE_BYTE_ARR: |
| throw new IllegalArgumentException("Byte arrays cannot be used as cache keys."); |
| |
| case CacheObject.TYPE_REGULAR: |
| return new KeyCacheObjectImpl(ctx.kernalContext().cacheObjects().unmarshal(ctx, bytes, null), bytes, -1); |
| } |
| |
| throw new IllegalArgumentException("Invalid object type: " + type); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public CacheObject toCacheObject(CacheObjectContext ctx, ByteBuffer buf) { |
| int len = buf.getInt(); |
| |
| assert len >= 0 : len; |
| |
| byte type = buf.get(); |
| |
| byte[] data = new byte[len]; |
| |
| buf.get(data); |
| |
| return toCacheObject(ctx, type, data); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IncompleteCacheObject toCacheObject(CacheObjectContext ctx, ByteBuffer buf, |
| @Nullable IncompleteCacheObject incompleteObj) { |
| if (incompleteObj == null) |
| incompleteObj = new IncompleteCacheObject(buf); |
| |
| if (incompleteObj.isReady()) |
| return incompleteObj; |
| |
| incompleteObj.readData(buf); |
| |
| if (incompleteObj.isReady()) |
| incompleteObj.object(toCacheObject(ctx, incompleteObj.type(), incompleteObj.data())); |
| |
| return incompleteObj; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IncompleteCacheObject toKeyCacheObject(CacheObjectContext ctx, ByteBuffer buf, |
| @Nullable IncompleteCacheObject incompleteObj) throws IgniteCheckedException { |
| if (incompleteObj == null) |
| incompleteObj = new IncompleteCacheObject(buf); |
| |
| if (incompleteObj.isReady()) |
| return incompleteObj; |
| |
| incompleteObj.readData(buf); |
| |
| if (incompleteObj.isReady()) |
| incompleteObj.object(toKeyCacheObject(ctx, incompleteObj.type(), incompleteObj.data())); |
| |
| return incompleteObj; |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public CacheObject toCacheObject(CacheObjectContext ctx, @Nullable Object obj, |
| boolean userObj) { |
| return toCacheObject(ctx, obj, userObj, false); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object unwrapTemporary(GridCacheContext ctx, Object obj) throws BinaryObjectException { |
| if (!ctx.cacheObjectContext().binaryEnabled()) |
| return obj; |
| |
| if (obj instanceof BinaryObjectOffheapImpl) |
| return ((BinaryObjectOffheapImpl)obj).heapCopy(); |
| |
| return obj; |
| } |
| |
| /** |
| * @param obj Object. |
| * @return Binary object. |
| * @throws IgniteException In case of error. |
| */ |
| @Nullable public Object toBinary(@Nullable Object obj, boolean failIfUnregistered) throws IgniteException { |
| if (obj == null) |
| return null; |
| |
| if (isBinaryObject(obj)) |
| return obj; |
| |
| return marshalToBinary(obj, failIfUnregistered); |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public IgniteNodeValidationResult validateNode(ClusterNode rmtNode, |
| DiscoveryDataBag.JoiningNodeDiscoveryData discoData |
| ) { |
| IgniteNodeValidationResult res; |
| |
| if (getBoolean(IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK) || !(marsh instanceof BinaryMarshaller)) |
| return null; |
| |
| if ((res = validateBinaryConfiguration(rmtNode)) != null) |
| return res; |
| |
| return validateBinaryMetadata(rmtNode.id(), (Map<Integer, BinaryMetadataHolder>)discoData.joiningNodeData()); |
| } |
| |
| /** */ |
| private IgniteNodeValidationResult validateBinaryConfiguration(ClusterNode rmtNode) { |
| Object rmtBinaryCfg = rmtNode.attribute(IgniteNodeAttributes.ATTR_BINARY_CONFIGURATION); |
| |
| ClusterNode locNode = ctx.discovery().localNode(); |
| |
| Object locBinaryCfg = locNode.attribute(IgniteNodeAttributes.ATTR_BINARY_CONFIGURATION); |
| |
| if (!F.eq(locBinaryCfg, rmtBinaryCfg)) { |
| String msg = "Local node's binary configuration is not equal to remote node's binary configuration " + |
| "[locNodeId=%s, rmtNodeId=%s, locBinaryCfg=%s, rmtBinaryCfg=%s]"; |
| |
| return new IgniteNodeValidationResult(rmtNode.id(), |
| String.format(msg, locNode.id(), rmtNode.id(), locBinaryCfg, rmtBinaryCfg), |
| String.format(msg, rmtNode.id(), locNode.id(), rmtBinaryCfg, locBinaryCfg)); |
| } |
| |
| return null; |
| } |
| |
| /** */ |
| private IgniteNodeValidationResult validateBinaryMetadata(UUID rmtNodeId, Map<Integer, BinaryMetadataHolder> newNodeMeta) { |
| if (newNodeMeta == null) |
| return null; |
| |
| for (Map.Entry<Integer, BinaryMetadataHolder> metaEntry : newNodeMeta.entrySet()) { |
| if (!metadataLocCache.containsKey(metaEntry.getKey())) |
| continue; |
| |
| BinaryMetadata locMeta = metadataLocCache.get(metaEntry.getKey()).metadata(); |
| BinaryMetadata rmtMeta = metaEntry.getValue().metadata(); |
| |
| if (locMeta == null || rmtMeta == null) |
| continue; |
| |
| try { |
| mergeMetadata(locMeta, rmtMeta); |
| } |
| catch (Exception e) { |
| String locMsg = "Exception was thrown when merging binary metadata from node %s: %s"; |
| |
| String rmtMsg = "Exception was thrown on coordinator when merging binary metadata from this node: %s"; |
| |
| return new IgniteNodeValidationResult(rmtNodeId, |
| String.format(locMsg, rmtNodeId.toString(), e.getMessage()), |
| String.format(rmtMsg, e.getMessage())); |
| } |
| } |
| |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public DiscoveryDataExchangeType discoveryDataType() { |
| return BINARY_PROC; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void collectGridNodeData(DiscoveryDataBag dataBag) { |
| if (!dataBag.commonDataCollectedFor(BINARY_PROC.ordinal())) { |
| Map<Integer, BinaryMetadataHolder> res = U.newHashMap(metadataLocCache.size()); |
| |
| for (Map.Entry<Integer,BinaryMetadataHolder> e : metadataLocCache.entrySet()) { |
| if (!e.getValue().removing()) |
| res.put(e.getKey(), e.getValue()); |
| } |
| |
| dataBag.addGridCommonData(BINARY_PROC.ordinal(), (Serializable) res); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) { |
| Map<Integer, BinaryMetadataHolder> res = U.newHashMap(metadataLocCache.size()); |
| |
| for (Map.Entry<Integer,BinaryMetadataHolder> e : metadataLocCache.entrySet()) |
| res.put(e.getKey(), e.getValue()); |
| |
| dataBag.addJoiningNodeData(BINARY_PROC.ordinal(), (Serializable) res); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onJoiningNodeDataReceived(DiscoveryDataBag.JoiningNodeDiscoveryData data) { |
| Map<Integer,BinaryMetadataHolder> newNodeMeta = (Map<Integer, BinaryMetadataHolder>) data.joiningNodeData(); |
| |
| if (newNodeMeta == null) |
| return; |
| |
| UUID joiningNode = data.joiningNodeId(); |
| |
| for (Map.Entry<Integer, BinaryMetadataHolder> metaEntry : newNodeMeta.entrySet()) { |
| if (metadataLocCache.containsKey(metaEntry.getKey())) { |
| BinaryMetadataHolder localMetaHolder = metadataLocCache.get(metaEntry.getKey()); |
| |
| BinaryMetadata newMeta = metaEntry.getValue().metadata(); |
| BinaryMetadata localMeta = localMetaHolder.metadata(); |
| |
| BinaryMetadata mergedMeta = mergeMetadata(localMeta, newMeta); |
| |
| if (mergedMeta != localMeta) { |
| //put mergedMeta to local cache and store to disk |
| U.log(log, |
| String.format("Newer version of existing BinaryMetadata[typeId=%d, typeName=%s] " + |
| "is received from node %s; updating it locally", |
| mergedMeta.typeId(), |
| mergedMeta.typeName(), |
| joiningNode)); |
| |
| metadataLocCache.put(metaEntry.getKey(), |
| new BinaryMetadataHolder(mergedMeta, |
| localMetaHolder.pendingVersion(), |
| localMetaHolder.acceptedVersion())); |
| |
| if (!ctx.clientNode()) |
| metadataFileStore.writeMetadata(mergedMeta); |
| } |
| } |
| else { |
| BinaryMetadataHolder newMetaHolder = metaEntry.getValue(); |
| BinaryMetadata newMeta = newMetaHolder.metadata(); |
| |
| U.log(log, |
| String.format("New BinaryMetadata[typeId=%d, typeName=%s] " + |
| "is received from node %s; adding it locally", |
| newMeta.typeId(), |
| newMeta.typeName(), |
| joiningNode)); |
| |
| metadataLocCache.put(metaEntry.getKey(), newMetaHolder); |
| |
| if (!ctx.clientNode()) |
| metadataFileStore.writeMetadata(newMeta); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onGridDataReceived(GridDiscoveryData data) { |
| Map<Integer, BinaryMetadataHolder> receivedData = (Map<Integer, BinaryMetadataHolder>) data.commonData(); |
| |
| if (receivedData != null) { |
| for (Map.Entry<Integer, BinaryMetadataHolder> e : receivedData.entrySet()) { |
| BinaryMetadataHolder holder = e.getValue(); |
| |
| BinaryMetadataHolder localHolder = new BinaryMetadataHolder(holder.metadata(), |
| holder.pendingVersion(), |
| holder.pendingVersion()); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Received metadata on join: " + localHolder); |
| |
| metadataLocCache.put(e.getKey(), localHolder); |
| |
| if (!ctx.clientNode()) |
| metadataFileStore.writeMetadata(holder.metadata()); |
| } |
| } |
| } |
| |
| /** |
| * Sets path to binary metadata store configured by user, should include binary_meta and consistentId |
| * @param binaryMetadataFileStoreDir path to binary_meta |
| */ |
| public void setBinaryMetadataFileStoreDir(@Nullable File binaryMetadataFileStoreDir) { |
| this.binaryMetadataFileStoreDir = binaryMetadataFileStoreDir; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void removeType(int typeId) { |
| BinaryMetadataHolder oldHld = metadataLocCache.get(typeId); |
| |
| if (oldHld == null) |
| throw new IgniteException("Failed to remove metadata, type not found: " + typeId); |
| |
| if (oldHld.removing()) |
| throw new IgniteException("Failed to remove metadata, type is being removed: " + typeId); |
| |
| if (!IgniteFeatures.allNodesSupports(ctx.discovery().allNodes(), IgniteFeatures.REMOVE_METADATA)) { |
| throw new IgniteException("Failed to remove metadata, " + |
| "all cluster nodes must support the remove type feature"); |
| } |
| |
| try { |
| GridFutureAdapter<MetadataUpdateResult> fut = transport.requestMetadataRemove(typeId); |
| |
| MetadataUpdateResult res = fut.get(); |
| |
| if (res.rejected()) |
| throw res.error(); |
| } |
| catch (IgniteCheckedException e) { |
| IgniteCheckedException ex = e; |
| |
| if (ctx.isStopping()) { |
| ex = new NodeStoppingException("Node is stopping."); |
| |
| ex.addSuppressed(e); |
| } |
| |
| throw new BinaryObjectException("Failed to remove metadata for type: " + typeId, ex); |
| } |
| } |
| |
| /** */ |
| @SuppressWarnings("PublicInnerClass") |
| public static class TestBinaryContext extends BinaryContext { |
| /** */ |
| private List<TestBinaryContextListener> listeners; |
| |
| /** |
| * @param metaHnd Meta handler. |
| * @param igniteCfg Ignite config. |
| * @param log Logger. |
| */ |
| public TestBinaryContext(BinaryMetadataHandler metaHnd, IgniteConfiguration igniteCfg, |
| IgniteLogger log) { |
| super(metaHnd, igniteCfg, log); |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public BinaryType metadata(int typeId) throws BinaryObjectException { |
| BinaryType metadata = super.metadata(typeId); |
| |
| if (listeners != null) { |
| for (TestBinaryContextListener listener : listeners) |
| listener.onAfterMetadataRequest(typeId, metadata); |
| } |
| |
| return metadata; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void updateMetadata(int typeId, BinaryMetadata meta, |
| boolean failIfUnregistered) throws BinaryObjectException { |
| if (listeners != null) { |
| for (TestBinaryContextListener listener : listeners) |
| listener.onBeforeMetadataUpdate(typeId, meta); |
| } |
| |
| super.updateMetadata(typeId, meta, failIfUnregistered); |
| } |
| |
| /** */ |
| public interface TestBinaryContextListener { |
| /** |
| * @param typeId Type id. |
| * @param type Type. |
| */ |
| void onAfterMetadataRequest(int typeId, BinaryType type); |
| |
| /** |
| * @param typeId Type id. |
| * @param metadata Metadata. |
| */ |
| void onBeforeMetadataUpdate(int typeId, BinaryMetadata metadata); |
| } |
| |
| /** |
| * @param lsnr Listener. |
| */ |
| public void addListener(TestBinaryContextListener lsnr) { |
| if (listeners == null) |
| listeners = new ArrayList<>(); |
| |
| if (!listeners.contains(lsnr)) |
| listeners.add(lsnr); |
| } |
| |
| /** */ |
| public void clearAllListener() { |
| if (listeners != null) |
| listeners.clear(); |
| } |
| } |
| } |
