From 9967ca1551dbf383210008eda646b94ce17ca7d0 Mon Sep 17 00:00:00 2001 From: pigxcloud <2270033969@qq.com> Date: Tue, 21 May 2019 15:31:57 +0800 Subject: [PATCH] seata --- pigx-upms/pigx-upms-biz/pom.xml | 5 + pigx-visual/pigx-seata-server/pom.xml | 48 ++ .../server/AbstractTCInboundHandler.java | 240 +++++++ .../src/main/java/io/seata/server/Server.java | 78 +++ .../java/io/seata/server/UUIDGenerator.java | 96 +++ .../io/seata/server/coordinator/Core.java | 55 ++ .../seata/server/coordinator/CoreFactory.java | 46 ++ .../coordinator/DefaultCoordinator.java | 362 +++++++++++ .../seata/server/coordinator/DefaultCore.java | 310 +++++++++ .../server/lock/DefaultLockManagerImpl.java | 186 ++++++ .../io/seata/server/lock/LockManager.java | 63 ++ .../seata/server/lock/LockManagerFactory.java | 46 ++ .../session/AbstractSessionManager.java | 191 ++++++ .../seata/server/session/BranchSession.java | 395 ++++++++++++ .../server/session/DefaultSessionManager.java | 65 ++ .../session/FileBasedSessionManager.java | 232 +++++++ .../seata/server/session/GlobalSession.java | 541 ++++++++++++++++ .../io/seata/server/session/Lockable.java | 42 ++ .../io/seata/server/session/Reloadable.java | 29 + .../server/session/SessionCondition.java | 55 ++ .../seata/server/session/SessionHelper.java | 121 ++++ .../seata/server/session/SessionHolder.java | 242 +++++++ .../server/session/SessionLifecycle.java | 89 +++ .../session/SessionLifecycleListener.java | 90 +++ .../seata/server/session/SessionManager.java | 109 ++++ .../store/FileTransactionStoreManager.java | 590 ++++++++++++++++++ .../io/seata/server/store/FlushDiskMode.java | 44 ++ .../seata/server/store/SessionStorable.java | 38 ++ .../io/seata/server/store/StoreConfig.java | 56 ++ .../server/store/TransactionStoreManager.java | 118 ++++ .../server/store/TransactionWriteStore.java | 129 ++++ .../src/main/resources/logback.xml | 54 ++ .../src/main/resources/registry.conf | 71 +++ pigx-visual/pom.xml | 1 + 34 files changed, 4837 insertions(+) create mode 100755 pigx-visual/pigx-seata-server/pom.xml create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java create mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java create mode 100755 pigx-visual/pigx-seata-server/src/main/resources/logback.xml create mode 100755 pigx-visual/pigx-seata-server/src/main/resources/registry.conf diff --git a/pigx-upms/pigx-upms-biz/pom.xml b/pigx-upms/pigx-upms-biz/pom.xml index bee4afd42..009eda897 100644 --- a/pigx-upms/pigx-upms-biz/pom.xml +++ b/pigx-upms/pigx-upms-biz/pom.xml @@ -92,6 +92,11 @@ mysql mysql-connector-java + + + org.springframework.cloud + spring-cloud-starter-alibaba-seata + org.springframework.boot diff --git a/pigx-visual/pigx-seata-server/pom.xml b/pigx-visual/pigx-seata-server/pom.xml new file mode 100755 index 000000000..1d9c33897 --- /dev/null +++ b/pigx-visual/pigx-seata-server/pom.xml @@ -0,0 +1,48 @@ + + + + + com.pig4cloud + pigx-visual + 3.0.0 + + + 4.0.0 + pigx-seata-server + jar + pigx-seata-server + + + + io.seata + seata-core + 0.5.2 + + + io.seata + seata-config-all + 0.5.2 + + + io.seata + seata-discovery-all + 0.5.2 + + + diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java new file mode 100755 index 000000000..d6655196e --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java @@ -0,0 +1,240 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server; + +import io.seata.common.XID; +import io.seata.core.exception.AbstractExceptionHandler; +import io.seata.core.exception.TransactionException; +import io.seata.core.model.GlobalStatus; +import io.seata.core.protocol.transaction.BranchRegisterRequest; +import io.seata.core.protocol.transaction.BranchRegisterResponse; +import io.seata.core.protocol.transaction.BranchReportRequest; +import io.seata.core.protocol.transaction.BranchReportResponse; +import io.seata.core.protocol.transaction.GlobalBeginRequest; +import io.seata.core.protocol.transaction.GlobalBeginResponse; +import io.seata.core.protocol.transaction.GlobalCommitRequest; +import io.seata.core.protocol.transaction.GlobalCommitResponse; +import io.seata.core.protocol.transaction.GlobalLockQueryRequest; +import io.seata.core.protocol.transaction.GlobalLockQueryResponse; +import io.seata.core.protocol.transaction.GlobalRollbackRequest; +import io.seata.core.protocol.transaction.GlobalRollbackResponse; +import io.seata.core.protocol.transaction.GlobalStatusRequest; +import io.seata.core.protocol.transaction.GlobalStatusResponse; +import io.seata.core.protocol.transaction.TCInboundHandler; +import io.seata.core.rpc.RpcContext; +import io.seata.server.session.GlobalSession; +import io.seata.server.session.SessionHolder; + +/** + * The type Abstract tc inbound handler. + * + * @author sharajava + */ +public abstract class AbstractTCInboundHandler extends AbstractExceptionHandler implements TCInboundHandler { + + @Override + public GlobalBeginResponse handle(GlobalBeginRequest request, final RpcContext rpcContext) { + GlobalBeginResponse response = new GlobalBeginResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(GlobalBeginRequest request, GlobalBeginResponse response) throws TransactionException { + doGlobalBegin(request, response, rpcContext); + } + }, request, response); + return response; + } + + /** + * Do global begin. + * + * @param request the request + * @param response the response + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doGlobalBegin(GlobalBeginRequest request, GlobalBeginResponse response, + RpcContext rpcContext) throws TransactionException; + + @Override + public GlobalCommitResponse handle(GlobalCommitRequest request, final RpcContext rpcContext) { + GlobalCommitResponse response = new GlobalCommitResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(GlobalCommitRequest request, GlobalCommitResponse response) + throws TransactionException { + doGlobalCommit(request, response, rpcContext); + } + }, request, response); + return response; + } + + /** + * Do global commit. + * + * @param request the request + * @param response the response + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doGlobalCommit(GlobalCommitRequest request, GlobalCommitResponse response, + RpcContext rpcContext) throws TransactionException; + + @Override + public GlobalRollbackResponse handle(GlobalRollbackRequest request, final RpcContext rpcContext) { + GlobalRollbackResponse response = new GlobalRollbackResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(GlobalRollbackRequest request, GlobalRollbackResponse response) + throws TransactionException { + doGlobalRollback(request, response, rpcContext); + } + + @Override + public void onTransactionException(GlobalRollbackRequest request, GlobalRollbackResponse response, + TransactionException tex) { + super.onTransactionException(request, response, tex); + GlobalSession globalSession = SessionHolder.findGlobalSession( + XID.getTransactionId(request.getXid())); + if (globalSession != null) { + response.setGlobalStatus(globalSession.getStatus()); + } else { + response.setGlobalStatus(GlobalStatus.Finished); + } + } + + @Override + public void onException(GlobalRollbackRequest request, GlobalRollbackResponse response, Exception rex) { + super.onException(request, response, rex); + GlobalSession globalSession = SessionHolder.findGlobalSession( + XID.getTransactionId(request.getXid())); + if (globalSession != null) { + response.setGlobalStatus(globalSession.getStatus()); + } else { + response.setGlobalStatus(GlobalStatus.Finished); + } + } + }, request, response); + return response; + } + + /** + * Do global rollback. + * + * @param request the request + * @param response the response + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doGlobalRollback(GlobalRollbackRequest request, GlobalRollbackResponse response, + RpcContext rpcContext) throws TransactionException; + + @Override + public BranchRegisterResponse handle(BranchRegisterRequest request, final RpcContext rpcContext) { + BranchRegisterResponse response = new BranchRegisterResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(BranchRegisterRequest request, BranchRegisterResponse response) + throws TransactionException { + doBranchRegister(request, response, rpcContext); + } + }, request, response); + return response; + } + + /** + * Do branch register. + * + * @param request the request + * @param response the response + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doBranchRegister(BranchRegisterRequest request, BranchRegisterResponse response, + RpcContext rpcContext) throws TransactionException; + + @Override + public BranchReportResponse handle(BranchReportRequest request, final RpcContext rpcContext) { + BranchReportResponse response = new BranchReportResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(BranchReportRequest request, BranchReportResponse response) + throws TransactionException { + doBranchReport(request, response, rpcContext); + } + }, request, response); + return response; + } + + /** + * Do branch report. + * + * @param request the request + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doBranchReport(BranchReportRequest request, BranchReportResponse response, + RpcContext rpcContext) + throws TransactionException; + + @Override + public GlobalLockQueryResponse handle(GlobalLockQueryRequest request, final RpcContext rpcContext) { + GlobalLockQueryResponse response = new GlobalLockQueryResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(GlobalLockQueryRequest request, GlobalLockQueryResponse response) + throws TransactionException { + doLockCheck(request, response, rpcContext); + } + }, request, response); + return response; + } + + /** + * Do lock check. + * + * @param request the request + * @param response the response + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doLockCheck(GlobalLockQueryRequest request, GlobalLockQueryResponse response, + RpcContext rpcContext) throws TransactionException; + + @Override + public GlobalStatusResponse handle(GlobalStatusRequest request, final RpcContext rpcContext) { + GlobalStatusResponse response = new GlobalStatusResponse(); + exceptionHandleTemplate(new AbstractCallback() { + @Override + public void execute(GlobalStatusRequest request, GlobalStatusResponse response) + throws TransactionException { + doGlobalStatus(request, response, rpcContext); + } + }, request, response); + return response; + } + + /** + * Do global status. + * + * @param request the request + * @param response the response + * @param rpcContext the rpc context + * @throws TransactionException the transaction exception + */ + protected abstract void doGlobalStatus(GlobalStatusRequest request, GlobalStatusResponse response, + RpcContext rpcContext) throws TransactionException; + +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java new file mode 100755 index 000000000..43df651ee --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java @@ -0,0 +1,78 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server; + +import io.seata.common.XID; +import io.seata.common.thread.NamedThreadFactory; +import io.seata.common.util.NetUtil; +import io.seata.core.rpc.netty.RpcServer; +import io.seata.core.rpc.netty.ShutdownHook; +import io.seata.server.coordinator.DefaultCoordinator; +import io.seata.server.session.SessionHolder; + +import java.io.IOException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * The type Server. + * + * @author jimin.jm @alibaba-inc.com + */ +public class Server { + + private static final int MIN_SERVER_POOL_SIZE = 10; + private static final int MAX_SERVER_POOL_SIZE = 50; + private static final int MAX_TASK_QUEUE_SIZE = 2000; + private static final int KEEP_ALIVE_TIME = 50; + private static final int SERVER_DEFAULT_PORT = 8091; + private static final ThreadPoolExecutor WORKING_THREADS = new ThreadPoolExecutor(MIN_SERVER_POOL_SIZE, + MAX_SERVER_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, + new LinkedBlockingQueue(MAX_TASK_QUEUE_SIZE), + new NamedThreadFactory("ServerHandlerThread", MAX_SERVER_POOL_SIZE), new ThreadPoolExecutor.CallerRunsPolicy()); + + /** + * The entry point of application. + * + * @param args the input arguments + * @throws IOException the io exception + */ + public static void main(String[] args) throws IOException { + RpcServer rpcServer = new RpcServer(WORKING_THREADS); + + rpcServer.setListenPort(SERVER_DEFAULT_PORT); + + //log store mode : file、db + String storeMode = "file"; + + UUIDGenerator.init(1); + SessionHolder.init(storeMode); + + DefaultCoordinator coordinator = new DefaultCoordinator(rpcServer); + coordinator.init(); + rpcServer.setHandler(coordinator); + // register ShutdownHook + ShutdownHook.getInstance().addDisposable(coordinator); + + XID.setIpAddress(NetUtil.getLocalIp()); + XID.setPort(rpcServer.getListenPort()); + + rpcServer.init(); + + System.exit(0); + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java new file mode 100755 index 000000000..efc55e555 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java @@ -0,0 +1,96 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server; + +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Calendar; +import java.util.Date; +import java.util.concurrent.atomic.AtomicLong; + +import io.seata.common.exception.ShouldNeverHappenException; + +/** + * The type Uuid generator. + * + * @author sharajava + */ +public class UUIDGenerator { + + private static final AtomicLong UUID = new AtomicLong(1000); + private static int serverNodeId = 1; + private static final long UUID_INTERNAL = 2000000000; + + /** + * Generate uuid long. + * + * @return the long + */ + public static long generateUUID() { + long id = UUID.incrementAndGet(); + if (id >= UUID_INTERNAL * (serverNodeId + 1)) { + synchronized (UUID) { + if (UUID.get() >= id) { + id -= UUID_INTERNAL; + UUID.set(id); + } + } + } + return id; + } + + /** + * Gets current uuid. + * + * @return the current uuid + */ + public static long getCurrentUUID() { + return UUID.get(); + } + + /** + * Sets uuid. + * + * @param expect the expect + * @param update the update + * @return the uuid + */ + public static boolean setUUID(long expect, long update) { + return UUID.compareAndSet(expect, update); + + } + + /** + * Init. + * + * @param serverNodeId the server node id + */ + public static void init(int serverNodeId) { + try { + UUIDGenerator.serverNodeId = serverNodeId; + UUID.set(UUID_INTERNAL * serverNodeId); + SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd"); + Calendar cal = Calendar.getInstance(); + Date date = format.parse("2019-01-01"); + cal.setTime(date); + long base = cal.getTimeInMillis(); + long current = System.currentTimeMillis(); + UUID.addAndGet((current - base) / 1000); + } catch (ParseException e) { + throw new ShouldNeverHappenException(e); + } + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java new file mode 100755 index 000000000..7d34dfee1 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java @@ -0,0 +1,55 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.coordinator; + +import io.seata.core.exception.TransactionException; +import io.seata.core.model.ResourceManagerInbound; +import io.seata.core.model.ResourceManagerOutbound; +import io.seata.core.model.TransactionManager; +import io.seata.server.session.GlobalSession; + +/** + * The interface Core. + * + * @author sharajava + */ +public interface Core extends TransactionManager, ResourceManagerOutbound { + + /** + * Sets resource manager inbound. + * + * @param resourceManagerInbound the resource manager inbound + */ + void setResourceManagerInbound(ResourceManagerInbound resourceManagerInbound); + + /** + * Do global commit. + * + * @param globalSession the global session + * @param retrying the retrying + * @throws TransactionException the transaction exception + */ + void doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException; + + /** + * Do global rollback. + * + * @param globalSession the global session + * @param retrying the retrying + * @throws TransactionException the transaction exception + */ + void doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException; +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java new file mode 100755 index 000000000..19e09f70e --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java @@ -0,0 +1,46 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.coordinator; + +/** + * The type Core factory. + * + * @author sharajava + */ +public class CoreFactory { + + private static class SingletonHolder { + private static Core INSTANCE = new DefaultCore(); + } + + /** + * Get core. + * + * @return the core + */ + public static final Core get() { + return SingletonHolder.INSTANCE; + } + + /** + * Just for test mocking + * + * @param core the core + */ + public static void set(Core core) { + SingletonHolder.INSTANCE = core; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java new file mode 100755 index 000000000..024f741a9 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java @@ -0,0 +1,362 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.coordinator; + +import io.seata.common.XID; +import io.seata.common.thread.NamedThreadFactory; +import io.seata.core.exception.TransactionException; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.BranchType; +import io.seata.core.model.GlobalStatus; +import io.seata.core.model.ResourceManagerInbound; +import io.seata.core.protocol.AbstractMessage; +import io.seata.core.protocol.AbstractResultMessage; +import io.seata.core.protocol.transaction.*; +import io.seata.core.rpc.Disposable; +import io.seata.core.rpc.RpcContext; +import io.seata.core.rpc.ServerMessageSender; +import io.seata.core.rpc.TransactionMessageHandler; +import io.seata.core.rpc.netty.RpcServer; +import io.seata.server.AbstractTCInboundHandler; +import io.seata.server.session.BranchSession; +import io.seata.server.session.GlobalSession; +import io.seata.server.session.SessionHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collection; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static io.seata.core.exception.TransactionExceptionCode.FailedToSendBranchCommitRequest; +import static io.seata.core.exception.TransactionExceptionCode.FailedToSendBranchRollbackRequest; + +/** + * The type Default coordinator. + * + * @author sharajava + */ +public class DefaultCoordinator extends AbstractTCInboundHandler + implements TransactionMessageHandler, ResourceManagerInbound, Disposable { + + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultCoordinator.class); + + private static final int TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS = 5000; + + private ServerMessageSender messageSender; + + private Core core = CoreFactory.get(); + + /** + * Instantiates a new Default coordinator. + * + * @param messageSender the message sender + */ + public DefaultCoordinator(ServerMessageSender messageSender) { + this.messageSender = messageSender; + core.setResourceManagerInbound(this); + } + + @Override + protected void doGlobalBegin(GlobalBeginRequest request, GlobalBeginResponse response, RpcContext rpcContext) + throws TransactionException { + response.setXid(core.begin(rpcContext.getApplicationId(), rpcContext.getTransactionServiceGroup(), + request.getTransactionName(), request.getTimeout())); + } + + @Override + protected void doGlobalCommit(GlobalCommitRequest request, GlobalCommitResponse response, RpcContext rpcContext) + throws TransactionException { + response.setGlobalStatus(core.commit(request.getXid())); + + } + + @Override + protected void doGlobalRollback(GlobalRollbackRequest request, GlobalRollbackResponse response, + RpcContext rpcContext) throws TransactionException { + response.setGlobalStatus(core.rollback(request.getXid())); + + } + + @Override + protected void doGlobalStatus(GlobalStatusRequest request, GlobalStatusResponse response, RpcContext rpcContext) + throws TransactionException { + response.setGlobalStatus(core.getStatus(request.getXid())); + } + + @Override + protected void doBranchRegister(BranchRegisterRequest request, BranchRegisterResponse response, + RpcContext rpcContext) throws TransactionException { + response.setBranchId( + core.branchRegister(request.getBranchType(), request.getResourceId(), rpcContext.getClientId(), + request.getXid(), request.getApplicationData(), request.getLockKey())); + + } + + @Override + protected void doBranchReport(BranchReportRequest request, BranchReportResponse response, RpcContext rpcContext) + throws TransactionException { + core.branchReport(request.getBranchType(), request.getXid(), request.getBranchId(), + request.getStatus(), + request.getApplicationData()); + + } + + @Override + protected void doLockCheck(GlobalLockQueryRequest request, GlobalLockQueryResponse response, RpcContext rpcContext) + throws TransactionException { + response.setLockable(core.lockQuery(request.getBranchType(), request.getResourceId(), + request.getXid(), request.getLockKey())); + } + + @Override + public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId, + String applicationData) + throws TransactionException { + try { + BranchCommitRequest request = new BranchCommitRequest(); + request.setXid(xid); + request.setBranchId(branchId); + request.setResourceId(resourceId); + request.setApplicationData(applicationData); + request.setBranchType(branchType); + + GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); + if(globalSession == null){ + return BranchStatus.PhaseTwo_Committed; + } + BranchSession branchSession = globalSession.getBranch(branchId); + + BranchCommitResponse response = (BranchCommitResponse)messageSender.sendSyncRequest(resourceId, + branchSession.getClientId(), request); + return response.getBranchStatus(); + } catch (IOException e) { + throw new TransactionException(FailedToSendBranchCommitRequest, branchId + "/" + xid, e); + } catch (TimeoutException e) { + throw new TransactionException(FailedToSendBranchCommitRequest, branchId + "/" + xid, e); + } + } + + @Override + public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId, + String applicationData) + throws TransactionException { + try { + BranchRollbackRequest + request = new BranchRollbackRequest(); + request.setXid(xid); + request.setBranchId(branchId); + request.setResourceId(resourceId); + request.setApplicationData(applicationData); + request.setBranchType(branchType); + + GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); + if(globalSession == null){ + return BranchStatus.PhaseTwo_Rollbacked; + } + BranchSession branchSession = globalSession.getBranch(branchId); + + BranchRollbackResponse response = (BranchRollbackResponse)messageSender.sendSyncRequest(resourceId, + branchSession.getClientId(), request); + return response.getBranchStatus(); + } catch (IOException e) { + throw new TransactionException(FailedToSendBranchRollbackRequest, branchId + "/" + xid, e); + } catch (TimeoutException e) { + throw new TransactionException(FailedToSendBranchRollbackRequest, branchId + "/" + xid, e); + } + } + + private void timeoutCheck() throws TransactionException { + Collection allSessions = SessionHolder.getRootSessionManager().allSessions(); + if (allSessions.size() > 0 && LOGGER.isDebugEnabled()) { + LOGGER.debug("Transaction Timeout Check Begin: " + allSessions.size()); + } + for (GlobalSession globalSession : allSessions) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug(globalSession.getTransactionId() + " " + globalSession.getStatus() + " " + + globalSession.getBeginTime() + " " + globalSession.getTimeout()); + } + boolean shouldTimeout = globalSession.lockAndExcute(() -> { + if (globalSession.getStatus() != GlobalStatus.Begin || !globalSession.isTimeout()) { + return false; + } + globalSession.close(); + globalSession.changeStatus(GlobalStatus.TimeoutRollbacking); + return true; + }); + if (!shouldTimeout) { + continue; + } + LOGGER.info( + "Global transaction[" + globalSession.getTransactionId() + "] is timeout and will be rolled back."); + + globalSession.addSessionLifecycleListener(SessionHolder.getRetryRollbackingSessionManager()); + SessionHolder.getRetryRollbackingSessionManager().addGlobalSession(globalSession); + + } + if (allSessions.size() > 0 && LOGGER.isDebugEnabled()) { + LOGGER.debug("Transaction Timeout Check End. "); + } + + } + + private void handleRetryRollbacking() { + Collection rollbackingSessions = SessionHolder.getRetryRollbackingSessionManager().allSessions(); + for (GlobalSession rollbackingSession : rollbackingSessions) { + try { + core.doGlobalRollback(rollbackingSession, true); + } catch (TransactionException ex) { + LOGGER.info("Failed to retry rollbacking [{}] {} {}", + rollbackingSession.getTransactionId(), ex.getCode(), ex.getMessage()); + } + } + } + + private void handleRetryCommitting() { + Collection committingSessions = SessionHolder.getRetryCommittingSessionManager().allSessions(); + for (GlobalSession committingSession : committingSessions) { + try { + core.doGlobalCommit(committingSession, true); + } catch (TransactionException ex) { + LOGGER.info("Failed to retry committing [{}] {} {}", + committingSession.getTransactionId(), ex.getCode(), ex.getMessage()); + } + } + } + + private void handleAsyncCommitting() { + Collection asyncCommittingSessions = SessionHolder.getAsyncCommittingSessionManager() + .allSessions(); + for (GlobalSession asyncCommittingSession : asyncCommittingSessions) { + try { + core.doGlobalCommit(asyncCommittingSession, true); + } catch (TransactionException ex) { + LOGGER.info("Failed to async committing [{}] {} {}", + asyncCommittingSession.getTransactionId(), ex.getCode(), ex.getMessage()); + } + } + } + + private ScheduledThreadPoolExecutor retryRollbacking = new ScheduledThreadPoolExecutor(1, + new NamedThreadFactory("RetryRollbacking", 1)); + + private ScheduledThreadPoolExecutor retryCommitting = new ScheduledThreadPoolExecutor(1, + new NamedThreadFactory("RetryCommitting", 1)); + + private ScheduledThreadPoolExecutor asyncCommitting = new ScheduledThreadPoolExecutor(1, + new NamedThreadFactory("AsyncCommitting", 1)); + + private ScheduledThreadPoolExecutor timeoutCheck = new ScheduledThreadPoolExecutor(1, + new NamedThreadFactory("TxTimeoutCheck", 1)); + + /** + * Init. + */ + public void init() { + retryRollbacking.scheduleAtFixedRate(new Runnable() { + + @Override + public void run() { + try { + handleRetryRollbacking(); + } catch (Exception e) { + LOGGER.info("Exception retry rollbacking ... ", e); + } + + } + }, 0, 5, TimeUnit.MILLISECONDS); + + retryCommitting.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + try { + handleRetryCommitting(); + } catch (Exception e) { + LOGGER.info("Exception retry committing ... ", e); + } + + } + }, 0, 5, TimeUnit.MILLISECONDS); + + asyncCommitting.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + try { + handleAsyncCommitting(); + } catch (Exception e) { + LOGGER.info("Exception async committing ... ", e); + } + + } + }, 0, 10, TimeUnit.MILLISECONDS); + + timeoutCheck.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + try { + timeoutCheck(); + } catch (Exception e) { + LOGGER.info("Exception timeout checking ... ", e); + } + + } + }, 0, 2, TimeUnit.MILLISECONDS); + } + + @Override + public AbstractResultMessage onRequest(AbstractMessage request, RpcContext context) { + if (!(request instanceof AbstractTransactionRequestToTC)) { + throw new IllegalArgumentException(); + } + AbstractTransactionRequestToTC transactionRequest = (AbstractTransactionRequestToTC)request; + transactionRequest.setTCInboundHandler(this); + + return transactionRequest.handle(context); + } + + @Override + public void onResponse(AbstractResultMessage response, RpcContext context) { + if (!(response instanceof AbstractTransactionResponse)) { + throw new IllegalArgumentException(); + } + + } + + @Override + public void destroy() { + // 1. first shutdown timed task + retryRollbacking.shutdown(); + retryCommitting.shutdown(); + asyncCommitting.shutdown(); + timeoutCheck.shutdown(); + try { + retryRollbacking.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); + retryCommitting.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); + asyncCommitting.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); + timeoutCheck.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); + } catch (InterruptedException ingore) { + + } + // 2. sencond close netty flow + if (messageSender instanceof RpcServer){ + ((RpcServer) messageSender).destroy(); + } + // 3. last destory SessionHolder + SessionHolder.destory(); + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java new file mode 100755 index 000000000..1ba36763f --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java @@ -0,0 +1,310 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.coordinator; + +import io.seata.common.XID; +import io.seata.core.exception.TransactionException; +import io.seata.core.exception.TransactionExceptionCode; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.BranchType; +import io.seata.core.model.GlobalStatus; +import io.seata.core.model.ResourceManagerInbound; +import io.seata.server.lock.LockManager; +import io.seata.server.lock.LockManagerFactory; +import io.seata.server.session.BranchSession; +import io.seata.server.session.GlobalSession; +import io.seata.server.session.SessionHelper; +import io.seata.server.session.SessionHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static io.seata.core.exception.TransactionExceptionCode.BranchTransactionNotExist; +import static io.seata.core.exception.TransactionExceptionCode.FailedToAddBranch; +import static io.seata.core.exception.TransactionExceptionCode.GlobalTransactionNotActive; +import static io.seata.core.exception.TransactionExceptionCode.GlobalTransactionStatusInvalid; +import static io.seata.core.exception.TransactionExceptionCode.LockKeyConflict; + +/** + * The type Default core. + * + * @author sharajava + */ +public class DefaultCore implements Core { + + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultCore.class); + + private LockManager lockManager = LockManagerFactory.get(); + + private ResourceManagerInbound resourceManagerInbound; + + @Override + public void setResourceManagerInbound(ResourceManagerInbound resourceManagerInbound) { + this.resourceManagerInbound = resourceManagerInbound; + } + + @Override + public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, + String applicationData, String lockKeys) throws TransactionException { + GlobalSession globalSession = assertGlobalSessionNotNull(XID.getTransactionId(xid)); + return globalSession.lockAndExcute(() -> { + if (!globalSession.isActive()) { + throw new TransactionException(GlobalTransactionNotActive, "Current Status: " + globalSession.getStatus()); + } + if (globalSession.getStatus() != GlobalStatus.Begin) { + throw new TransactionException(GlobalTransactionStatusInvalid, + globalSession.getStatus() + " while expecting " + GlobalStatus.Begin); + } + BranchSession branchSession = SessionHelper.newBranchByGlobal(globalSession, branchType, resourceId, + applicationData, lockKeys, clientId); + if (!branchSession.lock()) { + throw new TransactionException(LockKeyConflict); + } + try { + globalSession.addBranch(branchSession); + } catch (RuntimeException ex) { + throw new TransactionException(FailedToAddBranch); + } + return branchSession.getBranchId(); + }); + } + + private GlobalSession assertGlobalSessionNotNull(long transactionId) throws TransactionException { + GlobalSession globalSession = SessionHolder.findGlobalSession(transactionId); + if (globalSession == null) { + throw new TransactionException(TransactionExceptionCode.GlobalTransactionNotExist, "" + transactionId + ""); + } + return globalSession; + } + + + @Override + public void branchReport(BranchType branchType, String xid, long branchId, BranchStatus status, + String applicationData) throws TransactionException { + GlobalSession globalSession = assertGlobalSessionNotNull(XID.getTransactionId(xid)); + BranchSession branchSession = globalSession.getBranch(branchId); + if (branchSession == null) { + throw new TransactionException(BranchTransactionNotExist); + } + globalSession.changeBranchStatus(branchSession, status); + } + + @Override + public boolean lockQuery(BranchType branchType, String resourceId, String xid, String lockKeys) + throws TransactionException { + if (branchType == BranchType.AT) { + return lockManager.isLockable(XID.getTransactionId(xid), resourceId, lockKeys); + } else { + return true; + } + + } + + @Override + public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) + throws TransactionException { + GlobalSession session = GlobalSession.createGlobalSession( + applicationId, transactionServiceGroup, name, timeout); + session.addSessionLifecycleListener(SessionHolder.getRootSessionManager()); + + session.begin(); + + return XID.generateXID(session.getTransactionId()); + } + + @Override + public GlobalStatus commit(String xid) throws TransactionException { + GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); + if (globalSession == null) { + return GlobalStatus.Finished; + } + // just lock changeStatus + boolean shouldCommit = globalSession.lockAndExcute(() -> { + globalSession.closeAndClean(); // Highlight: Firstly, close the session, then no more branch can be registered. + if (globalSession.getStatus() == GlobalStatus.Begin) { + globalSession.changeStatus(GlobalStatus.Committing); + return true; + } + return false; + }); + if (!shouldCommit){ + return globalSession.getStatus(); + } + if (globalSession.canBeCommittedAsync()) { + asyncCommit(globalSession); + return GlobalStatus.Committed; + } else { + doGlobalCommit(globalSession, false); + } + return globalSession.getStatus(); + } + + @Override + public void doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException { + for (BranchSession branchSession : globalSession.getSortedBranches()) { + BranchStatus currentStatus = branchSession.getStatus(); + if (currentStatus == BranchStatus.PhaseOne_Failed) { + globalSession.removeBranch(branchSession); + continue; + } + try { + BranchStatus branchStatus = resourceManagerInbound.branchCommit(branchSession.getBranchType(), + XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(), + branchSession.getResourceId(), branchSession.getApplicationData()); + + switch (branchStatus) { + case PhaseTwo_Committed: + globalSession.removeBranch(branchSession); + continue; + case PhaseTwo_CommitFailed_Unretryable: + if (globalSession.canBeCommittedAsync()) { + LOGGER.error("By [{}], failed to commit branch {}", branchStatus, branchSession); + continue; + } else { + SessionHelper.endCommitFailed(globalSession); + LOGGER.error("Finally, failed to commit global[{}] since branch[{}] commit failed", + globalSession.getTransactionId(), branchSession.getBranchId()); + return; + } + default: + if (!retrying) { + queueToRetryCommit(globalSession); + return; + } + if (globalSession.canBeCommittedAsync()) { + LOGGER.error("By [{}], failed to commit branch {}", branchStatus, branchSession); + continue; + } else { + LOGGER.error( + "Failed to commit global[{}] since branch[{}] commit failed, will retry later.", + globalSession.getTransactionId(), branchSession.getBranchId()); + return; + } + + } + + } catch (Exception ex) { + LOGGER.error("Exception committing branch {}", branchSession, ex); + if (!retrying) { + queueToRetryCommit(globalSession); + throw new TransactionException(ex); + } + + } + + } + if (globalSession.hasBranch()) { + LOGGER.info("Global[{}] committing is NOT done.", globalSession.getTransactionId()); + return; + } + SessionHelper.endCommitted(globalSession); + LOGGER.info("Global[{}] committing is successfully done.", globalSession.getTransactionId()); + } + + private void asyncCommit(GlobalSession globalSession) throws TransactionException { + globalSession.addSessionLifecycleListener(SessionHolder.getAsyncCommittingSessionManager()); + SessionHolder.getAsyncCommittingSessionManager().addGlobalSession(globalSession); + globalSession.changeStatus(GlobalStatus.AsyncCommitting); + } + + private void queueToRetryCommit(GlobalSession globalSession) throws TransactionException { + globalSession.addSessionLifecycleListener(SessionHolder.getRetryCommittingSessionManager()); + SessionHolder.getRetryCommittingSessionManager().addGlobalSession(globalSession); + globalSession.changeStatus(GlobalStatus.CommitRetrying); + } + + private void queueToRetryRollback(GlobalSession globalSession) throws TransactionException { + globalSession.addSessionLifecycleListener(SessionHolder.getRetryRollbackingSessionManager()); + SessionHolder.getRetryRollbackingSessionManager().addGlobalSession(globalSession); + GlobalStatus currentStatus = globalSession.getStatus(); + if (SessionHelper.isTimeoutGlobalStatus(currentStatus)) { + globalSession.changeStatus(GlobalStatus.TimeoutRollbackRetrying); + } else { + globalSession.changeStatus(GlobalStatus.RollbackRetrying); + } + } + + @Override + public GlobalStatus rollback(String xid) throws TransactionException { + GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); + if (globalSession == null) { + return GlobalStatus.Finished; + } + // just lock changeStatus + boolean shouldRollBack = globalSession.lockAndExcute(() -> { + globalSession.close(); // Highlight: Firstly, close the session, then no more branch can be registered. + if (globalSession.getStatus() == GlobalStatus.Begin) { + globalSession.changeStatus(GlobalStatus.Rollbacking); + return true; + } + return false; + }); + if (!shouldRollBack){ + return globalSession.getStatus(); + } + + doGlobalRollback(globalSession, false); + return globalSession.getStatus(); + } + + @Override + public void doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException { + for (BranchSession branchSession : globalSession.getReverseSortedBranches()) { + BranchStatus currentBranchStatus = branchSession.getStatus(); + if (currentBranchStatus == BranchStatus.PhaseOne_Failed) { + globalSession.removeBranch(branchSession); + continue; + } + try { + BranchStatus branchStatus = resourceManagerInbound.branchRollback(branchSession.getBranchType(), + XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(), + branchSession.getResourceId(), branchSession.getApplicationData()); + + switch (branchStatus) { + case PhaseTwo_Rollbacked: + globalSession.removeBranch(branchSession); + LOGGER.error("Successfully rolled back branch " + branchSession); + continue; + case PhaseTwo_RollbackFailed_Unretryable: + SessionHelper.endRollbackFailed(globalSession); + LOGGER.error("Failed to rollback global[" + globalSession.getTransactionId() + "] since branch[" + + branchSession.getBranchId() + "] rollback failed"); + return; + default: + LOGGER.info("Failed to rollback branch " + branchSession); + if (!retrying) { + queueToRetryRollback(globalSession); + } + return; + + } + } catch (Exception ex) { + LOGGER.error("Exception rollbacking branch " + branchSession, ex); + if (!retrying) { + queueToRetryRollback(globalSession); + } + throw new TransactionException(ex); + } + + } + SessionHelper.endRollbacked(globalSession); + } + + @Override + public GlobalStatus getStatus(String xid) throws TransactionException { + GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); + return globalSession.getStatus(); + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java new file mode 100755 index 000000000..a0c576c84 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java @@ -0,0 +1,186 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.lock; + +import io.netty.util.internal.ConcurrentSet; +import io.seata.common.exception.ShouldNeverHappenException; +import io.seata.common.util.StringUtils; +import io.seata.core.exception.TransactionException; +import io.seata.server.session.BranchSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * The type Default lock manager. + * + * @author sharajava + */ +public class DefaultLockManagerImpl implements LockManager { + + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultLockManagerImpl.class); + + private static final int BUCKET_PER_TABLE = 128; + + private static final + ConcurrentHashMap>>> + LOCK_MAP = new ConcurrentHashMap>>>(); + + @Override + public boolean acquireLock(BranchSession branchSession) throws TransactionException { + String resourceId = branchSession.getResourceId(); + long transactionId = branchSession.getTransactionId(); + ConcurrentHashMap>> dbLockMap = LOCK_MAP.get(resourceId); + if (dbLockMap == null) { + LOCK_MAP.putIfAbsent(resourceId, + new ConcurrentHashMap>>()); + dbLockMap = LOCK_MAP.get(resourceId); + } + ConcurrentHashMap, Set> bucketHolder = branchSession.getLockHolder(); + + String lockKey = branchSession.getLockKey(); + if (StringUtils.isNullOrEmpty(lockKey)) { + return true; + } + + String[] tableGroupedLockKeys = lockKey.split(";"); + for (String tableGroupedLockKey : tableGroupedLockKeys) { + int idx = tableGroupedLockKey.indexOf(":"); + if (idx < 0) { + branchSession.unlock(); + throw new ShouldNeverHappenException("Wrong format of LOCK KEYS: " + branchSession.getLockKey()); + } + String tableName = tableGroupedLockKey.substring(0, idx); + String mergedPKs = tableGroupedLockKey.substring(idx + 1); + ConcurrentHashMap> tableLockMap = dbLockMap.get(tableName); + if (tableLockMap == null) { + dbLockMap.putIfAbsent(tableName, new ConcurrentHashMap>()); + tableLockMap = dbLockMap.get(tableName); + } + String[] pks = mergedPKs.split(","); + for (String pk : pks) { + int bucketId = pk.hashCode() % BUCKET_PER_TABLE; + Map bucketLockMap = tableLockMap.get(bucketId); + if (bucketLockMap == null) { + tableLockMap.putIfAbsent(bucketId, new HashMap()); + bucketLockMap = tableLockMap.get(bucketId); + } + synchronized (bucketLockMap) { + Long lockingTransactionId = bucketLockMap.get(pk); + if (lockingTransactionId == null) { + // No existing lock + bucketLockMap.put(pk, transactionId); + Set keysInHolder = bucketHolder.get(bucketLockMap); + if (keysInHolder == null) { + bucketHolder.putIfAbsent(bucketLockMap, new ConcurrentSet()); + keysInHolder = bucketHolder.get(bucketLockMap); + } + keysInHolder.add(pk); + + } else if (lockingTransactionId.longValue() == transactionId) { + // Locked by me + continue; + } else { + LOGGER.info( + "Global lock on [" + tableName + ":" + pk + "] is holding by " + lockingTransactionId); + branchSession.unlock(); // Release all acquired locks. + return false; + } + } + } + } + return true; + } + + @Override + public boolean isLockable(long transactionId, String resourceId, String lockKey) throws TransactionException { + ConcurrentHashMap>> dbLockMap = LOCK_MAP.get(resourceId); + if (dbLockMap == null) { + return true; + } + String[] tableGroupedLockKeys = lockKey.split(";"); + for (String tableGroupedLockKey : tableGroupedLockKeys) { + int idx = tableGroupedLockKey.indexOf(":"); + if (idx < 0) { + throw new ShouldNeverHappenException("Wrong format of LOCK KEYS: " + lockKey); + } + String tableName = tableGroupedLockKey.substring(0, idx); + String mergedPKs = tableGroupedLockKey.substring(idx + 1); + ConcurrentHashMap> tableLockMap = dbLockMap.get(tableName); + if (tableLockMap == null) { + continue; + } + String[] pks = mergedPKs.split(","); + for (String pk : pks) { + int bucketId = pk.hashCode() % BUCKET_PER_TABLE; + Map bucketLockMap = tableLockMap.get(bucketId); + if (bucketLockMap == null) { + continue; + } + Long lockingTransactionId = bucketLockMap.get(pk); + if (lockingTransactionId == null || lockingTransactionId.longValue() == transactionId) { + // Locked by me + continue; + } else { + LOGGER.info("Global lock on [" + tableName + ":" + pk + "] is holding by " + lockingTransactionId); + return false; + } + } + } + return true; + } + + @Override + public void cleanAllLocks() throws TransactionException { + LOCK_MAP.clear(); + } + + @Override + public boolean releaseLock(BranchSession branchSession) throws TransactionException { + ConcurrentHashMap, Set> lockHolder = branchSession.getLockHolder(); + if (lockHolder.size() == 0) { + return true; + } + Iterator, Set>> it = lockHolder.entrySet().iterator(); + while (it.hasNext()) { + Entry, Set> entry = it.next(); + Map bucket = entry.getKey(); + Set keys = entry.getValue(); + synchronized (bucket) { + for (String key : keys) { + Long v = bucket.get(key); + if (v == null) { + continue; + } + if (v.longValue() == branchSession.getTransactionId()) { + bucket.remove(key); + } + } + } + } + lockHolder.clear(); + return true; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java new file mode 100755 index 000000000..54bcef3c0 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java @@ -0,0 +1,63 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.lock; + +import io.seata.core.exception.TransactionException; +import io.seata.server.session.BranchSession; + +/** + * The interface Lock manager. + * + * @author sharajava + */ +public interface LockManager { + + /** + * Acquire lock boolean. + * + * @param branchSession the branch session + * @return the boolean + * @throws TransactionException the transaction exception + */ + boolean acquireLock(BranchSession branchSession) throws TransactionException; + + /** + * Is lockable boolean. + * + * @param transactionId the transaction id + * @param resourceId the resource id + * @param lockKey the lock key + * @return the boolean + * @throws TransactionException the transaction exception + */ + boolean isLockable(long transactionId, String resourceId, String lockKey) throws TransactionException; + + /** + * Clean all locks. + * + * @throws TransactionException the transaction exception + */ + void cleanAllLocks() throws TransactionException; + + /** + * Release lock boolean. + * + * @param branchSession the branch session + * @return the boolean + * @throws TransactionException the transaction exception + */ + boolean releaseLock(BranchSession branchSession) throws TransactionException; +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java new file mode 100755 index 000000000..3468cc894 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java @@ -0,0 +1,46 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.lock; + +/** + * The type Lock manager factory. + * + * @author sharajava + */ +public class LockManagerFactory { + + private static class SingletonHolder { + private static LockManager INSTANCE = new DefaultLockManagerImpl(); + } + + /** + * Get lock manager. + * + * @return the lock manager + */ + public static final LockManager get() { + return SingletonHolder.INSTANCE; + } + + /** + * Just for test mocking + * + * @param lockManager the lock manager + */ + public static void set(LockManager lockManager) { + SingletonHolder.INSTANCE = lockManager; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java new file mode 100755 index 000000000..d58f8b71d --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java @@ -0,0 +1,191 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.core.exception.TransactionException; +import io.seata.core.exception.TransactionExceptionCode; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.GlobalStatus; +import io.seata.server.store.SessionStorable; +import io.seata.server.store.TransactionStoreManager; +import io.seata.server.store.TransactionStoreManager.LogOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * The type Abstract session manager. + * + * @author sharajava + */ +public abstract class AbstractSessionManager implements SessionManager, SessionLifecycleListener { + + /** + * The constant LOGGER. + */ + protected static final Logger LOGGER = LoggerFactory.getLogger(AbstractSessionManager.class); + + /** + * The Session map. + */ + protected Map sessionMap = new ConcurrentHashMap<>(); + + /** + * The Transaction store manager. + */ + protected TransactionStoreManager transactionStoreManager; + + /** + * The Name. + */ + protected String name; + + /** + * Instantiates a new Abstract session manager. + * + * @param name the name + */ + public AbstractSessionManager(String name) { + this.name = name; + } + + @Override + public void addGlobalSession(GlobalSession session) throws TransactionException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("MANAGER[" + name + "] SESSION[" + session + "] " + LogOperation.GLOBAL_ADD); + } + writeSession(LogOperation.GLOBAL_ADD, session); + sessionMap.put(session.getTransactionId(), session); + + } + + + + @Override + public GlobalSession findGlobalSession(Long transactionId) { + return sessionMap.get(transactionId); + } + + @Override + public void updateGlobalSessionStatus(GlobalSession session, GlobalStatus status) throws TransactionException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("MANAGER[" + name + "] SESSION[" + session + "] " + LogOperation.GLOBAL_UPDATE); + } + writeSession(LogOperation.GLOBAL_UPDATE, session); + } + + @Override + public void removeGlobalSession(GlobalSession session) throws TransactionException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("MANAGER[" + name + "] SESSION[" + session + "] " + LogOperation.GLOBAL_REMOVE); + } + writeSession(LogOperation.GLOBAL_REMOVE, session); + sessionMap.remove(session.getTransactionId()); + + } + + @Override + public void addBranchSession(GlobalSession session, BranchSession branchSession) throws TransactionException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("MANAGER[" + name + "] SESSION[" + branchSession + "] " + LogOperation.BRANCH_ADD); + } + writeSession(LogOperation.BRANCH_ADD, branchSession); + } + + @Override + public void updateBranchSessionStatus(BranchSession branchSession, BranchStatus status) + throws TransactionException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("MANAGER[" + name + "] SESSION[" + branchSession + "] " + LogOperation.GLOBAL_ADD); + } + writeSession(LogOperation.BRANCH_UPDATE, branchSession); + } + + @Override + public void removeBranchSession(GlobalSession globalSession, BranchSession branchSession) + throws TransactionException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("MANAGER[" + name + "] SESSION[" + branchSession + "] " + LogOperation.GLOBAL_ADD); + } + writeSession(LogOperation.BRANCH_REMOVE, branchSession); + + } + + @Override + public Collection allSessions() { + return sessionMap.values(); + } + + @Override + public List findGlobalSessions(SessionCondition condition) { + List found = new ArrayList<>(); + for (GlobalSession globalSession : sessionMap.values()) { + if (System.currentTimeMillis() - globalSession.getBeginTime() > condition.getOverTimeAliveMills()) { + found.add(globalSession); + } + } + return found; + } + + @Override + public void onBegin(GlobalSession globalSession) throws TransactionException { + addGlobalSession(globalSession); + } + + @Override + public void onStatusChange(GlobalSession globalSession, GlobalStatus status) throws TransactionException { + updateGlobalSessionStatus(globalSession, status); + } + + @Override + public void onBranchStatusChange(GlobalSession globalSession, BranchSession branchSession, BranchStatus status) + throws TransactionException { + updateBranchSessionStatus(branchSession, status); + } + + @Override + public void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException { + addBranchSession(globalSession, branchSession); + } + + @Override + public void onRemoveBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException { + removeBranchSession(globalSession, branchSession); + } + + @Override + public void onClose(GlobalSession globalSession) throws TransactionException { + globalSession.setActive(false); + + } + + @Override + public void onEnd(GlobalSession globalSession) throws TransactionException { + removeGlobalSession(globalSession); + + } + + private void writeSession(LogOperation logOperation, SessionStorable sessionStorable) throws TransactionException{ + if (!transactionStoreManager.writeSession(logOperation, sessionStorable)) { + throw new TransactionException(TransactionExceptionCode.FailedWriteSession); + } + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java new file mode 100755 index 000000000..31e519612 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java @@ -0,0 +1,395 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import io.seata.common.util.CompressUtil; +import io.seata.core.exception.TransactionException; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.BranchType; +import io.seata.server.lock.LockManagerFactory; +import io.seata.server.store.SessionStorable; +import io.seata.server.store.StoreConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The type Branch session. + * + * @author sharajava + */ +public class BranchSession implements Lockable, Comparable, SessionStorable { + + private static final Logger LOGGER = LoggerFactory.getLogger(BranchSession.class); + + private static final int MAX_BRANCH_SESSION_SIZE = StoreConfig.getMaxBranchSessionSize(); + + private static ThreadLocal byteBufferThreadLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate( + MAX_BRANCH_SESSION_SIZE)); + + private long transactionId; + + private long branchId; + + private String resourceGroupId; + + private String resourceId; + + private String lockKey; + + private BranchType branchType; + + private BranchStatus status = BranchStatus.Unknown; + + private String clientId; + + private String applicationData; + + private ConcurrentHashMap, Set> lockHolder + = new ConcurrentHashMap, Set>(); + + /** + * Gets application data. + * + * @return the application data + */ + public String getApplicationData() { + return applicationData; + } + + /** + * Sets application data. + * + * @param applicationData the application data + */ + public void setApplicationData(String applicationData) { + this.applicationData = applicationData; + } + + /** + * Gets resource group id. + * + * @return the resource group id + */ + public String getResourceGroupId() { + return resourceGroupId; + } + + /** + * Sets resource group id. + * + * @param resourceGroupId the resource group id + */ + public void setResourceGroupId(String resourceGroupId) { + this.resourceGroupId = resourceGroupId; + } + + /** + * Gets client id. + * + * @return the client id + */ + public String getClientId() { + return clientId; + } + + /** + * Sets client id. + * + * @param clientId the client id + */ + public void setClientId(String clientId) { + this.clientId = clientId; + } + + /** + * Gets resource id. + * + * @return the resource id + */ + public String getResourceId() { + return resourceId; + } + + /** + * Sets resource id. + * + * @param resourceId the resource id + */ + public void setResourceId(String resourceId) { + this.resourceId = resourceId; + } + + /** + * Gets lock key. + * + * @return the lock key + */ + public String getLockKey() { + return lockKey; + } + + /** + * Sets lock key. + * + * @param lockKey the lock key + */ + public void setLockKey(String lockKey) { + this.lockKey = lockKey; + } + + /** + * Gets branch type. + * + * @return the branch type + */ + public BranchType getBranchType() { + return branchType; + } + + /** + * Sets branch type. + * + * @param branchType the branch type + */ + public void setBranchType(BranchType branchType) { + this.branchType = branchType; + } + + /** + * Gets status. + * + * @return the status + */ + public BranchStatus getStatus() { + return status; + } + + /** + * Sets status. + * + * @param status the status + */ + void setStatus(BranchStatus status) { + this.status = status; + } + + /** + * Gets transaction id. + * + * @return the transaction id + */ + public long getTransactionId() { + return transactionId; + } + + /** + * Sets transaction id. + * + * @param transactionId the transaction id + */ + public void setTransactionId(long transactionId) { + this.transactionId = transactionId; + } + + /** + * Gets branch id. + * + * @return the branch id + */ + public long getBranchId() { + return branchId; + } + + /** + * Sets branch id. + * + * @param branchId the branch id + */ + public void setBranchId(long branchId) { + this.branchId = branchId; + } + + @Override + public String toString() { + return "BR:" + branchId + "/" + transactionId; + } + + @Override + public int compareTo(BranchSession o) { + return this.branchId < o.branchId ? -1 : (this.branchId > o.branchId ? 1 : 0); + } + + /** + * Gets lock holder. + * + * @return the lock holder + */ + public ConcurrentHashMap, Set> getLockHolder() { + return lockHolder; + } + + @Override + public boolean lock() throws TransactionException { + return LockManagerFactory.get().acquireLock(this); + } + + @Override + public boolean unlock() throws TransactionException { + return LockManagerFactory.get().releaseLock(this); + } + + @Override + public byte[] encode() { + + byte[] resourceIdBytes = resourceId != null ? resourceId.getBytes() : null; + + byte[] lockKeyBytes = lockKey != null ? lockKey.getBytes() : null; + + byte[] clientIdBytes = clientId != null ? clientId.getBytes() : null; + + byte[] applicationDataBytes = applicationData != null ? applicationData.getBytes() : null; + + int size = calBranchSessionSize(resourceIdBytes, lockKeyBytes, clientIdBytes, applicationDataBytes); + + if (size > MAX_BRANCH_SESSION_SIZE) { + if (lockKeyBytes == null) { + throw new RuntimeException("branch session size exceeded, size : " + size + " maxBranchSessionSize : " + + MAX_BRANCH_SESSION_SIZE); + } + // try compress lockkey + try { + size -= lockKeyBytes.length; + lockKeyBytes = CompressUtil.compress(lockKeyBytes); + } catch (IOException e) { + LOGGER.error("compress lockKey error", e); + } finally { + size += lockKeyBytes.length; + } + + if (size > MAX_BRANCH_SESSION_SIZE) { + throw new RuntimeException( + "compress branch session size exceeded, compressSize : " + size + " maxBranchSessionSize : " + + MAX_BRANCH_SESSION_SIZE); + } + } + + ByteBuffer byteBuffer = byteBufferThreadLocal.get(); + //recycle + byteBuffer.clear(); + + byteBuffer.putLong(transactionId); + byteBuffer.putLong(branchId); + + if (null != resourceIdBytes) { + byteBuffer.putInt(resourceIdBytes.length); + byteBuffer.put(resourceIdBytes); + } else { + byteBuffer.putInt(0); + } + + if (null != lockKeyBytes) { + byteBuffer.putInt(lockKeyBytes.length); + byteBuffer.put(lockKeyBytes); + } else { + byteBuffer.putInt(0); + } + + if (null != clientIdBytes) { + byteBuffer.putShort((short)clientIdBytes.length); + byteBuffer.put(clientIdBytes); + } else { + byteBuffer.putShort((short)0); + } + + if (null != applicationDataBytes) { + byteBuffer.putInt(applicationDataBytes.length); + byteBuffer.put(applicationDataBytes); + } else { + byteBuffer.putInt(0); + } + + byteBuffer.put((byte)status.getCode()); + byteBuffer.flip(); + byte[] result = new byte[byteBuffer.limit()]; + byteBuffer.get(result); + return result; + } + + private int calBranchSessionSize(byte[] resourceIdBytes, byte[] lockKeyBytes, byte[] clientIdBytes, + byte[] applicationDataBytes) { + final int size = 8 // trascationId + + 8 // branchId + + 4 // resourceIdBytes.length + + 4 // lockKeyBytes.length + + 2 // clientIdBytes.length + + 4 // applicationDataBytes.length + + 1 // statusCode + + (resourceIdBytes == null ? 0 : resourceIdBytes.length) + + (lockKeyBytes == null ? 0 : lockKeyBytes.length) + + (clientIdBytes == null ? 0 : clientIdBytes.length) + + (applicationDataBytes == null ? 0 : applicationDataBytes.length); + return size; + } + + @Override + public void decode(byte[] a) { + ByteBuffer byteBuffer = ByteBuffer.wrap(a); + this.transactionId = byteBuffer.getLong(); + this.branchId = byteBuffer.getLong(); + int resourceLen = byteBuffer.getInt(); + if (resourceLen > 0) { + byte[] byResource = new byte[resourceLen]; + byteBuffer.get(byResource); + this.resourceId = new String(byResource); + } + int lockKeyLen = byteBuffer.getInt(); + if (lockKeyLen > 0) { + byte[] byLockKey = new byte[lockKeyLen]; + byteBuffer.get(byLockKey); + if (CompressUtil.isCompressData(byLockKey)) { + try { + this.lockKey = new String(CompressUtil.uncompress(byLockKey)); + } catch (IOException e) { + throw new RuntimeException("uncompress lockKey error", e); + } + } else { + this.lockKey = new String(byLockKey); + } + + } + short clientIdLen = byteBuffer.getShort(); + if (clientIdLen > 0) { + byte[] byClientId = new byte[clientIdLen]; + byteBuffer.get(byClientId); + this.clientId = new String(byClientId); + } + int applicationDataLen = byteBuffer.getInt(); + if (applicationDataLen > 0) { + byte[] byApplicationData = new byte[applicationDataLen]; + byteBuffer.get(byApplicationData); + this.applicationData = new String(byApplicationData); + } + this.status = BranchStatus.get(byteBuffer.get()); + + } + +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java new file mode 100755 index 000000000..ff24f7f6b --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java @@ -0,0 +1,65 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import java.util.List; + +import io.seata.server.store.SessionStorable; +import io.seata.server.store.TransactionStoreManager; +import io.seata.server.store.TransactionWriteStore; + +/** + * The type Default session manager. + * + * @author sharajava + */ +public class DefaultSessionManager extends AbstractSessionManager { + + /** + * Instantiates a new Default session manager. + * + * @param name the name + */ + public DefaultSessionManager(String name) { + super(name); + transactionStoreManager = new TransactionStoreManager() { + @Override + public boolean writeSession(LogOperation logOperation, SessionStorable session) { + return true; + } + + @Override + public void shutdown() { + + } + + @Override + public List readWriteStoreFromFile(int readSize, boolean isHistory) { + return null; + } + + @Override + public boolean hasRemaining(boolean isHistory) { + return false; + } + }; + } + + @Override + public void destroy() { + transactionStoreManager.shutdown(); + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java new file mode 100755 index 000000000..1684cccdf --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java @@ -0,0 +1,232 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import io.seata.common.exception.ShouldNeverHappenException; +import io.seata.config.ConfigurationFactory; +import io.seata.core.constants.ConfigurationKeys; +import io.seata.core.model.GlobalStatus; +import io.seata.server.UUIDGenerator; +import io.seata.server.store.FileTransactionStoreManager; +import io.seata.server.store.SessionStorable; +import io.seata.server.store.TransactionStoreManager; +import io.seata.server.store.TransactionWriteStore; + +/** + * The type File based session manager. + * + * @author jimin.jm @alibaba-inc.com + */ +public class FileBasedSessionManager extends AbstractSessionManager implements Reloadable { + + private static final int READ_SIZE = ConfigurationFactory.getInstance().getInt( + ConfigurationKeys.SERVICE_SESSION_RELOAD_READ_SIZE, 100); + + /** + * Instantiates a new File based session manager. + * + * @param name the name + * @param sessionStoreFilePath the session store file path + * @throws IOException the io exception + */ + public FileBasedSessionManager(String name, String sessionStoreFilePath) throws IOException { + super(name); + transactionStoreManager = new FileTransactionStoreManager(sessionStoreFilePath + File.separator + name, this); + } + + @Override + public void reload() { + restoreSessions(); + washSessions(); + } + + private void restoreSessions() { + Map unhandledBranchBuffer = new HashMap<>(); + + restoreSessions(true, unhandledBranchBuffer); + restoreSessions(false, unhandledBranchBuffer); + + if (!unhandledBranchBuffer.isEmpty()) { + unhandledBranchBuffer.values().forEach(branchSession -> { + long tid = branchSession.getTransactionId(); + long bid = branchSession.getBranchId(); + GlobalSession found = sessionMap.get(tid); + if (found == null) { + // Ignore + if (LOGGER.isInfoEnabled()) { + LOGGER.info("GlobalSession Does Not Exists For BranchSession [" + bid + "/" + tid + "]"); + } + } else { + BranchSession existingBranch = found.getBranch(branchSession.getBranchId()); + if (existingBranch == null) { + found.add(branchSession); + } else { + existingBranch.setStatus(branchSession.getStatus()); + } + } + + }); + } + } + + private void washSessions() { + if (sessionMap.size() > 0) { + Iterator> iterator = sessionMap.entrySet().iterator(); + while (iterator.hasNext()) { + GlobalSession globalSession = iterator.next().getValue(); + + GlobalStatus globalStatus = globalSession.getStatus(); + switch (globalStatus) { + case UnKnown: + case Committed: + case CommitFailed: + case Rollbacked: + case RollbackFailed: + case TimeoutRollbacked: + case TimeoutRollbackFailed: + case Finished: + // Remove all sessions finished + iterator.remove(); + break; + default: + break; + } + } + } + } + + private void restoreSessions(boolean isHistory, Map unhandledBranchBuffer) { + while (transactionStoreManager.hasRemaining(isHistory)) { + List stores = transactionStoreManager.readWriteStoreFromFile(READ_SIZE, isHistory); + restore(stores, unhandledBranchBuffer); + } + } + + private void restore(List stores, Map unhandledBranchSessions) { + long maxRecoverId = UUIDGenerator.getCurrentUUID(); + for (TransactionWriteStore store : stores) { + TransactionStoreManager.LogOperation logOperation = store.getOperate(); + SessionStorable sessionStorable = store.getSessionRequest(); + maxRecoverId = getMaxId(maxRecoverId, sessionStorable); + switch (logOperation) { + case GLOBAL_ADD: + case GLOBAL_UPDATE: { + GlobalSession globalSession = (GlobalSession)sessionStorable; + long tid = globalSession.getTransactionId(); + GlobalSession foundGlobalSession = sessionMap.get(tid); + if (foundGlobalSession == null) { + sessionMap.put(globalSession.getTransactionId(), globalSession); + } else { + foundGlobalSession.setStatus(globalSession.getStatus()); + } + break; + } + case GLOBAL_REMOVE: { + GlobalSession globalSession = (GlobalSession)sessionStorable; + long tid = globalSession.getTransactionId(); + if (sessionMap.remove(tid) == null) { + if (LOGGER.isInfoEnabled()) { + LOGGER.info("GlobalSession To Be Removed Does Not Exists [" + tid + "]"); + } + } + break; + } + case BRANCH_ADD: + case BRANCH_UPDATE: { + BranchSession branchSession = (BranchSession)sessionStorable; + long tid = branchSession.getTransactionId(); + GlobalSession foundGlobalSession = sessionMap.get(tid); + if (foundGlobalSession == null) { + unhandledBranchSessions.put(branchSession.getBranchId(), branchSession); + } else { + BranchSession existingBranch = foundGlobalSession.getBranch(branchSession.getBranchId()); + if (existingBranch == null) { + foundGlobalSession.add(branchSession); + } else { + existingBranch.setStatus(branchSession.getStatus()); + } + } + break; + } + case BRANCH_REMOVE: { + BranchSession branchSession = (BranchSession)sessionStorable; + long tid = branchSession.getTransactionId(); + long bid = branchSession.getBranchId(); + GlobalSession found = sessionMap.get(tid); + if (found == null) { + if (LOGGER.isInfoEnabled()) { + LOGGER.info( + "GlobalSession To Be Updated (Remove Branch) Does Not Exists [" + bid + "/" + tid + + "]"); + } + } else { + BranchSession theBranch = found.getBranch(bid); + if (theBranch == null) { + if (LOGGER.isInfoEnabled()) { + LOGGER.info("BranchSession To Be Updated Does Not Exists [" + bid + "/" + tid + "]"); + } + } else { + found.remove(theBranch); + } + } + break; + } + + default: + throw new ShouldNeverHappenException("Unknown Operation: " + logOperation); + + } + } + setMaxId(maxRecoverId); + + } + + private long getMaxId(long maxRecoverId, SessionStorable sessionStorable) { + long currentId = 0; + if (sessionStorable instanceof GlobalSession) { + currentId = ((GlobalSession)sessionStorable).getTransactionId(); + } else if (sessionStorable instanceof BranchSession) { + currentId = ((BranchSession)sessionStorable).getBranchId(); + } + + return maxRecoverId > currentId ? maxRecoverId : currentId; + } + + private void setMaxId(long maxRecoverId) { + long currentId; + // will be recover multi-thread later + while ((currentId = UUIDGenerator.getCurrentUUID()) < maxRecoverId) { + if (UUIDGenerator.setUUID(currentId, maxRecoverId)) { + break; + } + } + } + + private void restore(TransactionWriteStore store) { + } + + @Override + public void destroy() { + transactionStoreManager.shutdown(); + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java new file mode 100755 index 000000000..1c9f31d5b --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java @@ -0,0 +1,541 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.core.exception.TransactionException; +import io.seata.core.exception.TransactionExceptionCode; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.BranchType; +import io.seata.core.model.GlobalStatus; +import io.seata.server.UUIDGenerator; +import io.seata.server.store.SessionStorable; +import io.seata.server.store.StoreConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +/** + * The type Global session. + * + * @author sharajava + */ +public class GlobalSession implements SessionLifecycle, SessionStorable { + + private static final Logger LOGGER = LoggerFactory.getLogger(GlobalSession.class); + + private static final int MAX_GLOBAL_SESSION_SIZE = StoreConfig.getMaxGlobalSessionSize(); + + private static ThreadLocal byteBufferThreadLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate( + MAX_GLOBAL_SESSION_SIZE)); + + private long transactionId; + + private volatile GlobalStatus status; + + private String applicationId; + + private String transactionServiceGroup; + + private String transactionName; + + private int timeout; + + private long beginTime; + + private boolean active; + + private ArrayList branchSessions = new ArrayList<>(); + + private GlobalSessionLock globalSessionLock = new GlobalSessionLock(); + + /** + * Add boolean. + * + * @param branchSession the branch session + * @return the boolean + */ + public boolean add(BranchSession branchSession) { + return branchSessions.add(branchSession); + } + + /** + * Remove boolean. + * + * @param branchSession the branch session + * @return the boolean + */ + public boolean remove(BranchSession branchSession) { + return branchSessions.remove(branchSession); + } + + private ArrayList lifecycleListeners = new ArrayList<>(); + + /** + * Can be committed async boolean. + * + * @return the boolean + */ + public boolean canBeCommittedAsync() { + for (BranchSession branchSession : branchSessions) { + if (branchSession.getBranchType() == BranchType.TCC) { + return false; + } + } + return true; + } + + /** + * Is timeout boolean. + * + * @return the boolean + */ + public boolean isTimeout() { + return (System.currentTimeMillis() - beginTime) > timeout; + } + + @Override + public void begin() throws TransactionException { + this.status = GlobalStatus.Begin; + this.beginTime = System.currentTimeMillis(); + this.active = true; + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onBegin(this); + } + } + + @Override + public void changeStatus(GlobalStatus status) throws TransactionException { + this.status = status; + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onStatusChange(this, status); + } + + } + + @Override + public void changeBranchStatus(BranchSession branchSession, BranchStatus status) + throws TransactionException { + branchSession.setStatus(status); + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onBranchStatusChange(this, branchSession, status); + } + } + + @Override + public boolean isActive() { + return active; + } + + @Override + public void close() throws TransactionException { + if (active) { + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onClose(this); + } + } + } + + @Override + public void end() throws TransactionException { + // Clean locks first + clean(); + + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onEnd(this); + } + + } + + private void clean() throws TransactionException { + for (BranchSession branchSession : branchSessions) { + branchSession.unlock(); + } + + } + + /** + * Close and clean. + * + * @throws TransactionException the transaction exception + */ + public void closeAndClean() throws TransactionException { + close(); + clean(); + + } + + /** + * Add session lifecycle listener. + * + * @param sessionLifecycleListener the session lifecycle listener + */ + public void addSessionLifecycleListener(SessionLifecycleListener sessionLifecycleListener) { + lifecycleListeners.add(sessionLifecycleListener); + } + + /** + * Remove session lifecycle listener. + * + * @param sessionLifecycleListener the session lifecycle listener + */ + public void removeSessionLifecycleListener(SessionLifecycleListener sessionLifecycleListener) { + lifecycleListeners.remove(sessionLifecycleListener); + } + + @Override + public void addBranch(BranchSession branchSession) throws TransactionException { + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onAddBranch(this, branchSession); + } + branchSession.setStatus(BranchStatus.Registered); + add(branchSession); + } + + @Override + public void removeBranch(BranchSession branchSession) throws TransactionException { + for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { + lifecycleListener.onRemoveBranch(this, branchSession); + } + branchSession.unlock(); + remove(branchSession); + } + + /** + * Gets branch. + * + * @param branchId the branch id + * @return the branch + */ + public BranchSession getBranch(long branchId) { + synchronized (branchSessions) { + for (BranchSession branchSession : branchSessions) { + if (branchSession.getBranchId() == branchId) { + return branchSession; + } + } + + return null; + } + + } + + /** + * Gets sorted branches. + * + * @return the sorted branches + */ + public ArrayList getSortedBranches() { + ArrayList sorted = new ArrayList(); + sorted.addAll(branchSessions); + return sorted; + } + + /** + * Gets reverse sorted branches. + * + * @return the reverse sorted branches + */ + public ArrayList getReverseSortedBranches() { + ArrayList reversed = new ArrayList(); + reversed.addAll(branchSessions); + Collections.reverse(reversed); + return reversed; + } + + /** + * Instantiates a new Global session. + */ + public GlobalSession() {} + + /** + * Instantiates a new Global session. + * + * @param applicationId the application id + * @param transactionServiceGroup the transaction service group + * @param transactionName the transaction name + * @param timeout the timeout + */ + public GlobalSession(String applicationId, String transactionServiceGroup, String transactionName, int timeout) { + this.transactionId = UUIDGenerator.generateUUID(); + this.status = GlobalStatus.Begin; + + this.applicationId = applicationId; + this.transactionServiceGroup = transactionServiceGroup; + this.transactionName = transactionName; + this.timeout = timeout; + } + + /** + * Gets transaction id. + * + * @return the transaction id + */ + public long getTransactionId() { + return transactionId; + } + + /** + * Gets status. + * + * @return the status + */ + public GlobalStatus getStatus() { + return status; + } + + /** + * Sets status. + * + * @param status the status + */ + void setStatus(GlobalStatus status) { + this.status = status; + } + + /** + * Gets application id. + * + * @return the application id + */ + public String getApplicationId() { + return applicationId; + } + + /** + * Gets transaction service group. + * + * @return the transaction service group + */ + public String getTransactionServiceGroup() { + return transactionServiceGroup; + } + + /** + * Gets transaction name. + * + * @return the transaction name + */ + public String getTransactionName() { + return transactionName; + } + + /** + * Gets timeout. + * + * @return the timeout + */ + public int getTimeout() { + return timeout; + } + + /** + * Gets begin time. + * + * @return the begin time + */ + public long getBeginTime() { + return beginTime; + } + + + /** + * Create global session global session. + * + * @param applicationId the application id + * @param txServiceGroup the tx service group + * @param txName the tx name + * @param timeout the timeout + * @return the global session + */ + public static GlobalSession createGlobalSession(String applicationId, String txServiceGroup, String txName, + int timeout) { + GlobalSession session = new GlobalSession(applicationId, txServiceGroup, txName, timeout); + return session; + } + + /** + * Sets active. + * + * @param active the active + */ + public void setActive(boolean active) { + this.active = active; + } + + @Override + public byte[] encode() { + + byte[] byApplicationIdBytes = applicationId != null ? applicationId.getBytes() : null; + + byte[] byServiceGroupBytes = transactionServiceGroup != null ? transactionServiceGroup.getBytes() : null; + + byte[] byTxNameBytes = transactionName != null ? transactionName.getBytes() : null; + + int size = calGlobalSessionSize(byApplicationIdBytes, byServiceGroupBytes, byTxNameBytes); + + if (size > MAX_GLOBAL_SESSION_SIZE) { + throw new RuntimeException("global session size exceeded, size : " + size + " maxBranchSessionSize : " + + MAX_GLOBAL_SESSION_SIZE); + } + ByteBuffer byteBuffer = byteBufferThreadLocal.get(); + //recycle + byteBuffer.clear(); + + byteBuffer.putLong(transactionId); + byteBuffer.putInt(timeout); + if (null != byApplicationIdBytes) { + byteBuffer.putShort((short)byApplicationIdBytes.length); + byteBuffer.put(byApplicationIdBytes); + } else { + byteBuffer.putShort((short)0); + } + if (null != byServiceGroupBytes) { + byteBuffer.putShort((short)byServiceGroupBytes.length); + byteBuffer.put(byServiceGroupBytes); + } else { + byteBuffer.putShort((short)0); + } + if (null != byTxNameBytes) { + byteBuffer.putShort((short)byTxNameBytes.length); + byteBuffer.put(byTxNameBytes); + } else { + byteBuffer.putShort((short)0); + } + byteBuffer.putLong(beginTime); + byteBuffer.put((byte)status.getCode()); + byteBuffer.flip(); + byte[] result = new byte[byteBuffer.limit()]; + byteBuffer.get(result); + return result; + } + + private int calGlobalSessionSize(byte[] byApplicationIdBytes, byte[] byServiceGroupBytes, byte[] byTxNameBytes) { + final int size = 8 // trascationId + + 4 // timeout + + 2 // byApplicationIdBytes.length + + 2 // byServiceGroupBytes.length + + 2 // byTxNameBytes.length + + 8 // beginTime + + 1 // statusCode + + (byApplicationIdBytes == null ? 0 : byApplicationIdBytes.length) + + (byServiceGroupBytes == null ? 0 : byServiceGroupBytes.length) + + (byTxNameBytes == null ? 0 : byTxNameBytes.length); + return size; + } + + @Override + public void decode(byte[] a) { + ByteBuffer byteBuffer = ByteBuffer.wrap(a); + this.transactionId = byteBuffer.getLong(); + this.timeout = byteBuffer.getInt(); + short applicationIdLen = byteBuffer.getShort(); + if (applicationIdLen > 0) { + byte[] byApplicationId = new byte[applicationIdLen]; + byteBuffer.get(byApplicationId); + this.applicationId = new String(byApplicationId); + } + short serviceGroupLen = byteBuffer.getShort(); + if (serviceGroupLen > 0) { + byte[] byServiceGroup = new byte[serviceGroupLen]; + byteBuffer.get(byServiceGroup); + this.transactionServiceGroup = new String(byServiceGroup); + } + short txNameLen = byteBuffer.getShort(); + if (txNameLen > 0) { + byte[] byTxName = new byte[txNameLen]; + byteBuffer.get(byTxName); + this.transactionName = new String(byTxName); + } + this.beginTime = byteBuffer.getLong(); + this.status = GlobalStatus.get(byteBuffer.get()); + } + + /** + * Has branch boolean. + * + * @return the boolean + */ + public boolean hasBranch() { + return branchSessions.size() > 0; + } + + public void lock() throws TransactionException { + globalSessionLock.lock(); + } + + public void unlock(){ + globalSessionLock.unlock(); + } + + public void lockAndExcute(LockRunnable excuteRunnable) throws TransactionException{ + this.lock(); + try { + excuteRunnable.run(); + }finally { + this.unlock(); + } + } + + public T lockAndExcute(LockCallable lockCallable) throws TransactionException{ + this.lock(); + try { + return lockCallable.call(); + }finally { + this.unlock(); + } + } + + private class GlobalSessionLock { + + private Lock globalSessionLock = new ReentrantLock(); + + private static final int GLOBAL_SESSOION_LOCK_TIME_OUT_MILLS = 2 * 1000; + + public void lock() throws TransactionException { + try { + if (globalSessionLock.tryLock(GLOBAL_SESSOION_LOCK_TIME_OUT_MILLS, TimeUnit.MILLISECONDS)){ + return; + } + } catch (InterruptedException e) { + LOGGER.error("Interrupted error", e); + } + throw new TransactionException(TransactionExceptionCode.FailedLockGlobalTranscation); + } + + + public void unlock() { + globalSessionLock.unlock(); + } + + } + + @FunctionalInterface + public interface LockRunnable { + + void run() throws TransactionException; + } + @FunctionalInterface + public interface LockCallable { + + V call() throws TransactionException; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java new file mode 100755 index 000000000..d023e632f --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java @@ -0,0 +1,42 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.core.exception.TransactionException; + +/** + * The interface Lockable. + * + * @author sharajava + */ +public interface Lockable { + + /** + * Lock boolean. + * + * @return the boolean + * @throws TransactionException the transaction exception + */ + boolean lock() throws TransactionException; + + /** + * Unlock boolean. + * + * @return the boolean + * @throws TransactionException the transaction exception + */ + boolean unlock() throws TransactionException; +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java new file mode 100755 index 000000000..5701c69d6 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java @@ -0,0 +1,29 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +/** + * Service contains states which can be reloaded. + * + * @author sharajava + */ +public interface Reloadable { + + /** + * Reload states. + */ + void reload(); +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java new file mode 100755 index 000000000..4063d7aa3 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java @@ -0,0 +1,55 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +/** + * The type Session condition. + * + * @author jimin.jm @alibaba-inc.com + * @date 2018 /12/13 + */ +public class SessionCondition { + + private long overTimeAliveMills; + + /** + * Instantiates a new Session condition. + * + * @param overTimeAliveMills the over time alive mills + */ + public SessionCondition( long overTimeAliveMills) { + this.overTimeAliveMills = overTimeAliveMills; + } + + + /** + * Gets over time alive mills. + * + * @return the over time alive mills + */ + public long getOverTimeAliveMills() { + return overTimeAliveMills; + } + + /** + * Sets over time alive mills. + * + * @param overTimeAliveMills the over time alive mills + */ + public void setOverTimeAliveMills(long overTimeAliveMills) { + this.overTimeAliveMills = overTimeAliveMills; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java new file mode 100755 index 000000000..b3295d24e --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java @@ -0,0 +1,121 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.core.exception.TransactionException; +import io.seata.core.model.BranchType; +import io.seata.core.model.GlobalStatus; +import io.seata.server.UUIDGenerator; + +/** + * The type Session helper. + * + * @author sharajava + */ +public class SessionHelper { + + private SessionHelper() {} + + public static BranchSession newBranchByGlobal(GlobalSession globalSession, BranchType branchType, String resourceId, String lockKeys, String clientId) { + return newBranchByGlobal(globalSession, branchType, resourceId, null, lockKeys, clientId); + } + + /** + * New branch by global branch session. + * + * @param globalSession the global session + * @param branchType the branch type + * @param resourceId the resource id + * @param lockKeys the lock keys + * @param clientId the client id + * @return the branch session + */ + public static BranchSession newBranchByGlobal(GlobalSession globalSession, BranchType branchType, String resourceId, + String applicationData, String lockKeys, String clientId) { + BranchSession branchSession = new BranchSession(); + + branchSession.setTransactionId(globalSession.getTransactionId()); + branchSession.setBranchId(UUIDGenerator.generateUUID()); + branchSession.setBranchType(branchType); + branchSession.setResourceId(resourceId); + branchSession.setLockKey(lockKeys); + branchSession.setClientId(clientId); + branchSession.setApplicationData(applicationData); + + return branchSession; + } + + /** + * End committed. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + public static void endCommitted(GlobalSession globalSession) throws TransactionException { + globalSession.changeStatus(GlobalStatus.Committed); + globalSession.end(); + } + + /** + * End commit failed. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + public static void endCommitFailed(GlobalSession globalSession) throws TransactionException { + globalSession.changeStatus(GlobalStatus.CommitFailed); + globalSession.end(); + } + + /** + * End rollbacked. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + public static void endRollbacked(GlobalSession globalSession) throws TransactionException { + GlobalStatus currentStatus = globalSession.getStatus(); + if (isTimeoutGlobalStatus(currentStatus)) { + globalSession.changeStatus(GlobalStatus.TimeoutRollbacked); + } else { + globalSession.changeStatus(GlobalStatus.Rollbacked); + } + globalSession.end(); + } + + /** + * End rollback failed. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + public static void endRollbackFailed(GlobalSession globalSession) throws TransactionException { + GlobalStatus currentStatus = globalSession.getStatus(); + if (isTimeoutGlobalStatus(currentStatus)) { + globalSession.changeStatus(GlobalStatus.TimeoutRollbackFailed); + } else { + globalSession.changeStatus(GlobalStatus.RollbackFailed); + } + globalSession.end(); + } + + public static boolean isTimeoutGlobalStatus(GlobalStatus status) { + return status == GlobalStatus.TimeoutRollbacked + || status == GlobalStatus.TimeoutRollbackFailed + || status == GlobalStatus.TimeoutRollbacking + || status == GlobalStatus.TimeoutRollbackRetrying; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java new file mode 100755 index 000000000..a46e7001d --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java @@ -0,0 +1,242 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.common.exception.ShouldNeverHappenException; +import io.seata.common.exception.StoreException; +import io.seata.common.util.StringUtils; +import io.seata.config.Configuration; +import io.seata.config.ConfigurationFactory; +import io.seata.core.constants.ConfigurationKeys; +import io.seata.core.exception.TransactionException; +import io.seata.core.model.GlobalStatus; +import io.seata.core.store.StoreMode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; + +/** + * The type Session holder. + * + * @author sharajava + */ +public class SessionHolder { + + private static final Logger LOGGER = LoggerFactory.getLogger(SessionHolder.class); + + /** + * The constant CONFIG. + */ + protected static final Configuration CONFIG = ConfigurationFactory.getInstance(); + + /** + * The constant ROOT_SESSION_MANAGER_NAME. + */ + public static final String ROOT_SESSION_MANAGER_NAME = "root.data"; + /** + * The constant ASYNC_COMMITTING_SESSION_MANAGER_NAME. + */ + public static final String ASYNC_COMMITTING_SESSION_MANAGER_NAME = "async.commit.data"; + /** + * The constant RETRY_COMMITTING_SESSION_MANAGER_NAME. + */ + public static final String RETRY_COMMITTING_SESSION_MANAGER_NAME = "retry.commit.data"; + /** + * The constant RETRY_ROLLBACKING_SESSION_MANAGER_NAME. + */ + public static final String RETRY_ROLLBACKING_SESSION_MANAGER_NAME = "retry.rollback.data"; + + private static SessionManager ROOT_SESSION_MANAGER; + private static SessionManager ASYNC_COMMITTING_SESSION_MANAGER; + private static SessionManager RETRY_COMMITTING_SESSION_MANAGER; + private static SessionManager RETRY_ROLLBACKING_SESSION_MANAGER; + + /** + * Init. + * + * @param mode the store mode: file、db + * @throws IOException the io exception + */ + public static void init(String mode) throws IOException { + if (StringUtils.isBlank(mode)) { + //use default + mode = CONFIG.getConfig(ConfigurationKeys.STORE_MODE); + } + //the store mode + StoreMode storeMode = StoreMode.valueof(mode); + if (StoreMode.DB.equals(storeMode)) { + //TODO database store + + } else if (StoreMode.FILE.equals(storeMode)) { + //file store + String sessionStorePath = CONFIG.getConfig(ConfigurationKeys.STORE_FILE_DIR); + if (sessionStorePath == null) { + throw new StoreException("the {store.file.dir} is empty."); + } + ROOT_SESSION_MANAGER = new FileBasedSessionManager(ROOT_SESSION_MANAGER_NAME, sessionStorePath); + ASYNC_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(ASYNC_COMMITTING_SESSION_MANAGER_NAME); + RETRY_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(RETRY_COMMITTING_SESSION_MANAGER_NAME); + RETRY_ROLLBACKING_SESSION_MANAGER = new DefaultSessionManager(RETRY_ROLLBACKING_SESSION_MANAGER_NAME); + } else { + //unknown store + throw new IllegalArgumentException("unknown store mode:" + mode); + } + + if (ROOT_SESSION_MANAGER instanceof Reloadable) { + ((Reloadable) ROOT_SESSION_MANAGER).reload(); + + Collection reloadedSessions = ROOT_SESSION_MANAGER.allSessions(); + if (reloadedSessions != null && !reloadedSessions.isEmpty()) { + reloadedSessions.forEach(globalSession -> { + GlobalStatus globalStatus = globalSession.getStatus(); + switch (globalStatus) { + case UnKnown: + case Committed: + case CommitFailed: + case Rollbacked: + case RollbackFailed: + case TimeoutRollbacked: + case TimeoutRollbackFailed: + case Finished: + throw new ShouldNeverHappenException("Reloaded Session should NOT be " + globalStatus); + case AsyncCommitting: + try { + globalSession.addSessionLifecycleListener(getAsyncCommittingSessionManager()); + getAsyncCommittingSessionManager().addGlobalSession(globalSession); + } catch (TransactionException e) { + throw new ShouldNeverHappenException(e); + } + break; + default: { + ArrayList branchSessions = globalSession.getSortedBranches(); + // Lock + branchSessions.forEach(branchSession -> { + try { + branchSession.lock(); + } catch (TransactionException e) { + throw new ShouldNeverHappenException(e); + } + }); + + switch (globalStatus) { + case Committing: + case CommitRetrying: + try { + globalSession.addSessionLifecycleListener( + getRetryCommittingSessionManager()); + getRetryCommittingSessionManager().addGlobalSession(globalSession); + } catch (TransactionException e) { + throw new ShouldNeverHappenException(e); + } + break; + case Rollbacking: + case RollbackRetrying: + case TimeoutRollbacking: + case TimeoutRollbackRetrying: + try { + globalSession.addSessionLifecycleListener( + getRetryRollbackingSessionManager()); + getRetryRollbackingSessionManager().addGlobalSession(globalSession); + } catch (TransactionException e) { + throw new ShouldNeverHappenException(e); + } + break; + case Begin: + globalSession.setActive(true); + break; + default: + throw new ShouldNeverHappenException("NOT properly handled " + globalStatus); + } + + break; + + } + } + + }); + } + } + + } + + /** + * Gets root session manager. + * + * @return the root session manager + */ + public static final SessionManager getRootSessionManager() { + if (ROOT_SESSION_MANAGER == null) { + throw new ShouldNeverHappenException("SessionManager is NOT init!"); + } + return ROOT_SESSION_MANAGER; + } + + /** + * Gets async committing session manager. + * + * @return the async committing session manager + */ + public static final SessionManager getAsyncCommittingSessionManager() { + if (ASYNC_COMMITTING_SESSION_MANAGER == null) { + throw new ShouldNeverHappenException("SessionManager is NOT init!"); + } + return ASYNC_COMMITTING_SESSION_MANAGER; + } + + /** + * Gets retry committing session manager. + * + * @return the retry committing session manager + */ + public static final SessionManager getRetryCommittingSessionManager() { + if (RETRY_COMMITTING_SESSION_MANAGER == null) { + throw new ShouldNeverHappenException("SessionManager is NOT init!"); + } + return RETRY_COMMITTING_SESSION_MANAGER; + } + + /** + * Gets retry rollbacking session manager. + * + * @return the retry rollbacking session manager + */ + public static final SessionManager getRetryRollbackingSessionManager() { + if (RETRY_ROLLBACKING_SESSION_MANAGER == null) { + throw new ShouldNeverHappenException("SessionManager is NOT init!"); + } + return RETRY_ROLLBACKING_SESSION_MANAGER; + } + + /** + * Find global session. + * + * @param transactionId the transaction id + * @return the global session + */ + public static GlobalSession findGlobalSession(Long transactionId) { + return getRootSessionManager().findGlobalSession(transactionId); + } + + public static void destory() { + ROOT_SESSION_MANAGER.destroy(); + ASYNC_COMMITTING_SESSION_MANAGER.destroy(); + RETRY_COMMITTING_SESSION_MANAGER.destroy(); + RETRY_ROLLBACKING_SESSION_MANAGER.destroy(); + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java new file mode 100755 index 000000000..9c46a2fd7 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java @@ -0,0 +1,89 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.core.exception.TransactionException; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.GlobalStatus; + +/** + * The interface Session lifecycle. + * + * @author sharajava + */ +public interface SessionLifecycle { + + /** + * Begin. + * + * @throws TransactionException the transaction exception + */ + void begin() throws TransactionException; + + /** + * Change status. + * + * @param status the status + * @throws TransactionException the transaction exception + */ + void changeStatus(GlobalStatus status) throws TransactionException; + + /** + * Change branch status. + * + * @param branchSession the branch session + * @param status the status + * @throws TransactionException the transaction exception + */ + void changeBranchStatus(BranchSession branchSession, BranchStatus status) throws TransactionException; + + /** + * Add branch. + * + * @param branchSession the branch session + * @throws TransactionException the transaction exception + */ + void addBranch(BranchSession branchSession) throws TransactionException; + + /** + * Remove branch. + * + * @param branchSession the branch session + * @throws TransactionException the transaction exception + */ + void removeBranch(BranchSession branchSession) throws TransactionException; + + /** + * Is active boolean. + * + * @return the boolean + */ + boolean isActive(); + + /** + * Close. + * + * @throws TransactionException the transaction exception + */ + void close() throws TransactionException; + + /** + * End. + * + * @throws TransactionException the transaction exception + */ + void end() throws TransactionException; +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java new file mode 100755 index 000000000..54c415e92 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java @@ -0,0 +1,90 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import io.seata.core.exception.TransactionException; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.GlobalStatus; + +/** + * The interface Session lifecycle listener. + * + * @author sharajava + */ +public interface SessionLifecycleListener { + + /** + * On begin. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + void onBegin(GlobalSession globalSession) throws TransactionException; + + /** + * On status change. + * + * @param globalSession the global session + * @param status the status + * @throws TransactionException the transaction exception + */ + void onStatusChange(GlobalSession globalSession, GlobalStatus status) throws TransactionException; + + /** + * On branch status change. + * + * @param globalSession the global session + * @param branchSession the branch session + * @param status the status + * @throws TransactionException the transaction exception + */ + void onBranchStatusChange(GlobalSession globalSession, BranchSession branchSession, BranchStatus status) + throws TransactionException; + + /** + * On add branch. + * + * @param globalSession the global session + * @param branchSession the branch session + * @throws TransactionException the transaction exception + */ + void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException; + + /** + * On remove branch. + * + * @param globalSession the global session + * @param branchSession the branch session + * @throws TransactionException the transaction exception + */ + void onRemoveBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException; + + /** + * On close. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + void onClose(GlobalSession globalSession) throws TransactionException; + + /** + * On end. + * + * @param globalSession the global session + * @throws TransactionException the transaction exception + */ + void onEnd(GlobalSession globalSession) throws TransactionException; +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java new file mode 100755 index 000000000..2883d4404 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java @@ -0,0 +1,109 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.session; + +import java.util.Collection; +import java.util.List; + +import io.seata.core.exception.TransactionException; +import io.seata.core.model.BranchStatus; +import io.seata.core.model.GlobalStatus; +import io.seata.core.rpc.Disposable; + +/** + * The interface Session manager. + * + * @author sharajava + */ +public interface SessionManager extends SessionLifecycleListener, Disposable { + + /** + * Add global session. + * + * @param session the session + * @throws TransactionException the transaction exception + */ + void addGlobalSession(GlobalSession session) throws TransactionException; + + /** + * Find global session global session. + * + * @param transactionId the transaction id + * @return the global session + * @throws TransactionException the transaction exception + */ + GlobalSession findGlobalSession(Long transactionId); + + /** + * Update global session status. + * + * @param session the session + * @param status the status + * @throws TransactionException the transaction exception + */ + void updateGlobalSessionStatus(GlobalSession session, GlobalStatus status) throws TransactionException; + + /** + * Remove global session. + * + * @param session the session + * @throws TransactionException the transaction exception + */ + void removeGlobalSession(GlobalSession session) throws TransactionException; + + /** + * Add branch session. + * + * @param globalSession the global session + * @param session the session + * @throws TransactionException the transaction exception + */ + void addBranchSession(GlobalSession globalSession, BranchSession session) throws TransactionException; + + /** + * Update branch session status. + * + * @param session the session + * @param status the status + * @throws TransactionException the transaction exception + */ + void updateBranchSessionStatus(BranchSession session, BranchStatus status) throws TransactionException; + + /** + * Remove branch session. + * + * @param globalSession the global session + * @param session the session + * @throws TransactionException the transaction exception + */ + void removeBranchSession(GlobalSession globalSession, BranchSession session) throws TransactionException; + + /** + * All sessions collection. + * + * @return the collection + */ + Collection allSessions(); + + /** + * Find global sessions list. + * + * @param condition the condition + * @return the list + */ + List findGlobalSessions(SessionCondition condition); + +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java new file mode 100755 index 000000000..9b3a32229 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java @@ -0,0 +1,590 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.store; + +import io.seata.common.thread.NamedThreadFactory; +import io.seata.common.util.CollectionUtils; +import io.seata.server.session.BranchSession; +import io.seata.server.session.GlobalSession; +import io.seata.server.session.SessionCondition; +import io.seata.server.session.SessionManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; + +/** + * The type File transaction store manager. + * + * @author jimin.jm @alibaba-inc.com + */ +public class FileTransactionStoreManager implements TransactionStoreManager { + + private static final Logger LOGGER = LoggerFactory.getLogger(FileTransactionStoreManager.class); + + private static final int MAX_THREAD_WRITE = 1; + + private ExecutorService fileWriteExecutor; + + private volatile boolean stopping = false; + + private static final int MAX_SHUTDOWN_RETRY = 3; + + private static final int SHUTDOWN_CHECK_INTERNAL = 1 * 1000; + + private static final int MAX_WRITE_RETRY = 5; + + private static final String HIS_DATA_FILENAME_POSTFIX = ".1"; + + private static final AtomicLong FILE_TRX_NUM = new AtomicLong(0); + + private static final AtomicLong FILE_FLUSH_NUM = new AtomicLong(0); + + private static final int MARK_SIZE = 4; + + private static final int MAX_WAIT_TIME_MILLS = 2 * 1000; + + private static final int MAX_FLUSH_TIME_MILLS = 2 * 1000; + + private static final int MAX_FLUSH_NUM = 10; + + private static int PER_FILE_BLOCK_SIZE = 65535 * 8; + + private static long MAX_TRX_TIMEOUT_MILLS = 30 * 60 * 1000; + + private static volatile long trxStartTimeMills = System.currentTimeMillis(); + + private File currDataFile; + + private RandomAccessFile currRaf; + + private FileChannel currFileChannel; + + private long recoverCurrOffset = 0; + + private long recoverHisOffset = 0; + + private SessionManager sessionManager; + + private String currFullFileName; + + private String hisFullFileName; + + private WriteDataFileRunnable writeDataFileRunnable; + + private ReentrantLock writeSessionLock = new ReentrantLock(); + + private volatile long lastModifiedTime; + + private static final int MAX_WRITE_BUFFER_SIZE = StoreConfig.getFileWriteBufferCacheSize(); + + private final ByteBuffer writeBuffer = ByteBuffer.allocateDirect(MAX_WRITE_BUFFER_SIZE); + + private static final FlushDiskMode FLUSH_DISK_MODE = StoreConfig.getFlushDiskMode(); + + private static final int MAX_WAIT_FOR_FLUSH_TIME_MILLS = 2 * 1000; + + private static final int INT_BYTE_SIZE = 4; + + /** + * Instantiates a new File transaction store manager. + * + * @param fullFileName the dir path + * @param sessionManager the session manager + * @throws IOException the io exception + */ + public FileTransactionStoreManager(String fullFileName, SessionManager sessionManager) throws IOException { + initFile(fullFileName); + fileWriteExecutor = + new ThreadPoolExecutor(MAX_THREAD_WRITE, MAX_THREAD_WRITE, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, + new LinkedBlockingQueue(), + new NamedThreadFactory("fileTransactionStore", MAX_THREAD_WRITE, true)); + writeDataFileRunnable = new WriteDataFileRunnable(); + fileWriteExecutor.submit(writeDataFileRunnable); + this.sessionManager = sessionManager; + } + + private void initFile(String fullFileName) throws IOException { + this.currFullFileName = fullFileName; + this.hisFullFileName = fullFileName + HIS_DATA_FILENAME_POSTFIX; + try { + currDataFile = new File(currFullFileName); + if (!currDataFile.exists()) { + //create parent dir first + if (currDataFile.getParentFile() != null && !currDataFile.getParentFile().exists()) { + currDataFile.getParentFile().mkdirs(); + } + currDataFile.createNewFile(); + trxStartTimeMills = System.currentTimeMillis(); + } else { + trxStartTimeMills = currDataFile.lastModified(); + } + lastModifiedTime = System.currentTimeMillis(); + currRaf = new RandomAccessFile(currDataFile, "rw"); + currRaf.seek(currDataFile.length()); + currFileChannel = currRaf.getChannel(); + } catch (IOException exx) { + LOGGER.error("init file error," + exx.getMessage()); + throw exx; + } + } + + @Override + public boolean writeSession(LogOperation logOperation, SessionStorable session) { + writeSessionLock.lock(); + long curFileTrxNum; + try { + if (!writeDataFile(new TransactionWriteStore(session, logOperation).encode())) { + return false; + } + lastModifiedTime = System.currentTimeMillis(); + curFileTrxNum = FILE_TRX_NUM.incrementAndGet(); + if (curFileTrxNum % PER_FILE_BLOCK_SIZE == 0 && + (System.currentTimeMillis() - trxStartTimeMills) > MAX_TRX_TIMEOUT_MILLS) { + return saveHistory(); + } + } catch (Exception exx) { + LOGGER.error("writeSession error," + exx.getMessage()); + return false; + } finally { + writeSessionLock.unlock(); + } + flushDisk(curFileTrxNum, currFileChannel); + return true; + } + + private void flushDisk(long curFileNum, FileChannel currFileChannel) { + + if (FLUSH_DISK_MODE == FlushDiskMode.SYNC_MODEL) { + SyncFlushRequest syncFlushRequest = new SyncFlushRequest(curFileNum, currFileChannel); + writeDataFileRunnable.putRequest(syncFlushRequest); + syncFlushRequest.waitForFlush(MAX_WAIT_FOR_FLUSH_TIME_MILLS); + } else { + writeDataFileRunnable.putRequest(new AsyncFlushRequest(curFileNum, currFileChannel)); + } + } + + /** + * get all overTimeSessionStorables + * merge write file + * + * @throws IOException + */ + private boolean saveHistory() throws IOException { + boolean result; + try { + result = findTimeoutAndSave(); + writeDataFileRunnable.putRequest(new CloseFileRequest(currFileChannel, currRaf)); + Files.move(currDataFile.toPath(), new File(hisFullFileName).toPath(), StandardCopyOption.REPLACE_EXISTING); + } catch (IOException exx) { + LOGGER.error("save history data file error," + exx.getMessage()); + result = false; + } finally { + initFile(currFullFileName); + } + return result; + } + + private boolean findTimeoutAndSave() throws IOException { + List globalSessionsOverMaxTimeout = + sessionManager.findGlobalSessions(new SessionCondition(MAX_TRX_TIMEOUT_MILLS)); + if (CollectionUtils.isEmpty(globalSessionsOverMaxTimeout)) { + return true; + } + List listBytes = new ArrayList<>(); + int totalSize = 0; + // 1. find all data and merge + for (GlobalSession globalSession : globalSessionsOverMaxTimeout) { + TransactionWriteStore globalWriteStore = new TransactionWriteStore(globalSession, LogOperation.GLOBAL_ADD); + byte[] data = globalWriteStore.encode(); + listBytes.add(data); + totalSize += data.length + INT_BYTE_SIZE; + List branchSessIonsOverMaXTimeout = globalSession.getSortedBranches(); + if (null != branchSessIonsOverMaXTimeout) { + for (BranchSession branchSession : branchSessIonsOverMaXTimeout) { + TransactionWriteStore branchWriteStore = + new TransactionWriteStore(branchSession, LogOperation.BRANCH_ADD); + data = branchWriteStore.encode(); + listBytes.add(data); + totalSize += data.length + INT_BYTE_SIZE; + } + } + } + // 2. batch write + ByteBuffer byteBuffer = ByteBuffer.allocateDirect(totalSize); + for (byte[] bytes : listBytes) { + byteBuffer.putInt(bytes.length); + byteBuffer.put(bytes); + } + if (writeDataFileByBuffer(byteBuffer)) { + currFileChannel.force(false); + return true; + } + return false; + } + + @Override + public void shutdown() { + if (null != fileWriteExecutor) { + fileWriteExecutor.shutdown(); + stopping = true; + int retry = 0; + while (!fileWriteExecutor.isTerminated() && retry < MAX_SHUTDOWN_RETRY) { + ++retry; + try { + Thread.sleep(SHUTDOWN_CHECK_INTERNAL); + } catch (InterruptedException exx) { + } + } + if (retry >= MAX_SHUTDOWN_RETRY) { + fileWriteExecutor.shutdownNow(); + } + } + try { + currFileChannel.force(true); + } catch (IOException e) { + LOGGER.error("filechannel force error", e); + } + closeFile(currRaf); + } + + @Override + public List readWriteStoreFromFile(int readSize, boolean isHistory) { + File file = null; + long currentOffset = 0; + if (isHistory) { + file = new File(hisFullFileName); + currentOffset = recoverHisOffset; + } else { + file = new File(currFullFileName); + currentOffset = recoverCurrOffset; + } + if (file.exists()) { + return parseDataFile(file, readSize, currentOffset); + } + return null; + } + + @Override + public boolean hasRemaining(boolean isHistory) { + File file = null; + RandomAccessFile raf = null; + long currentOffset = 0; + if (isHistory) { + file = new File(hisFullFileName); + currentOffset = recoverHisOffset; + } else { + file = new File(currFullFileName); + currentOffset = recoverCurrOffset; + } + try { + raf = new RandomAccessFile(file, "r"); + return currentOffset < raf.length(); + + } catch (IOException exx) { + } finally { + closeFile(raf); + } + return false; + } + + private List parseDataFile(File file, int readSize, long currentOffset) { + List transactionWriteStores = new ArrayList<>(readSize); + RandomAccessFile raf = null; + FileChannel fileChannel = null; + try { + raf = new RandomAccessFile(file, "r"); + raf.seek(currentOffset); + fileChannel = raf.getChannel(); + fileChannel.position(currentOffset); + long size = raf.length(); + ByteBuffer buffSize = ByteBuffer.allocate(MARK_SIZE); + while (fileChannel.position() < size) { + try { + buffSize.clear(); + int avilReadSize = fileChannel.read(buffSize); + if (avilReadSize != MARK_SIZE) { + break; + } + buffSize.flip(); + int bodySize = buffSize.getInt(); + byte[] byBody = new byte[bodySize]; + ByteBuffer buffBody = ByteBuffer.wrap(byBody); + avilReadSize = fileChannel.read(buffBody); + if (avilReadSize != bodySize) { + break; + } + TransactionWriteStore writeStore = new TransactionWriteStore(); + writeStore.decode(byBody); + transactionWriteStores.add(writeStore); + if (transactionWriteStores.size() == readSize) { + break; + } + } catch (Exception ex) { + LOGGER.error("decode data file error:" + ex.getMessage()); + break; + } + } + return transactionWriteStores; + } catch (IOException exx) { + LOGGER.error("parse data file error:" + exx.getMessage() + ",file:" + file.getName()); + return null; + } finally { + try { + if (null != fileChannel) { + if (isHisFile(file)) { + recoverHisOffset = fileChannel.position(); + } else { + recoverCurrOffset = fileChannel.position(); + } + } + closeFile(raf); + } catch (IOException exx) { + LOGGER.error("file close error," + exx.getMessage()); + } + } + + } + + private boolean isHisFile(File file) { + + return file.getName().endsWith(HIS_DATA_FILENAME_POSTFIX); + } + + private void closeFile(RandomAccessFile raf) { + try { + if (null != raf) { + raf.close(); + raf = null; + } + } catch (IOException exx) { + LOGGER.error("file close error," + exx.getMessage()); + } + } + + private boolean writeDataFile(byte[] bs) { + if (bs == null) { + return false; + } + ByteBuffer byteBuffer = null; + + if (bs.length > MAX_WRITE_BUFFER_SIZE) { + //allocateNew + byteBuffer = ByteBuffer.allocateDirect(bs.length); + } else { + byteBuffer = writeBuffer; + //recycle + byteBuffer.clear(); + } + + byteBuffer.putInt(bs.length); + byteBuffer.put(bs); + return writeDataFileByBuffer(byteBuffer); + } + + private boolean writeDataFileByBuffer(ByteBuffer byteBuffer) { + byteBuffer.flip(); + for (int retry = 0; retry < MAX_WRITE_RETRY; retry++) { + try { + while (byteBuffer.hasRemaining()) { + currFileChannel.write(byteBuffer); + } + return true; + } catch (IOException exx) { + LOGGER.error("write data file error:" + exx.getMessage()); + } + } + LOGGER.error("write dataFile failed,retry more than :" + MAX_WRITE_RETRY); + return false; + } + + interface StoreRequest { + + } + abstract class AbstractFlushRequest implements StoreRequest{ + private final long curFileTrxNum; + + private final FileChannel curFileChannel; + + protected AbstractFlushRequest(long curFileTrxNum, FileChannel curFileChannel) { + this.curFileTrxNum = curFileTrxNum; + this.curFileChannel = curFileChannel; + } + public long getCurFileTrxNum() { + return curFileTrxNum; + } + + public FileChannel getCurFileChannel() { + return curFileChannel; + } + } + class SyncFlushRequest extends AbstractFlushRequest { + + private final CountDownLatch countDownLatch = new CountDownLatch(1); + + public SyncFlushRequest(long curFileTrxNum, FileChannel curFileChannel) { + super(curFileTrxNum, curFileChannel); + } + + + public void wakeupCustomer() { + this.countDownLatch.countDown(); + } + + public void waitForFlush(long timeout) { + try { + this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + LOGGER.error("Interrupted", e); + } + } + } + + class AsyncFlushRequest extends AbstractFlushRequest { + + public AsyncFlushRequest(long curFileTrxNum, FileChannel curFileChannel) { + super(curFileTrxNum, curFileChannel); + } + + } + + class CloseFileRequest implements StoreRequest { + + private FileChannel fileChannel; + + private RandomAccessFile file; + + public CloseFileRequest(FileChannel fileChannel, RandomAccessFile file) { + this.fileChannel = fileChannel; + this.file = file; + } + + public FileChannel getFileChannel() { + return fileChannel; + } + + public RandomAccessFile getFile() { + return file; + } + } + + /** + * The type Write data file runnable. + */ + class WriteDataFileRunnable implements Runnable { + + private LinkedBlockingQueue storeRequests = new LinkedBlockingQueue<>(); + + + public void putRequest(final StoreRequest request) { + storeRequests.add(request); + } + + + @Override + public void run() { + while (!stopping) { + try { + StoreRequest storeRequest = storeRequests.poll(MAX_WAIT_TIME_MILLS, TimeUnit.MILLISECONDS); + handleStoreRequest(storeRequest); + } catch (Exception exx) { + LOGGER.error("write file error", exx.getMessage()); + } + } + handleRestRequest(); + } + + /** + * handle the rest requests when stopping is true + */ + private void handleRestRequest() { + int remainNums = storeRequests.size(); + for (int i = 0; i < remainNums; i++) { + handleStoreRequest(storeRequests.poll()); + } + } + + private void handleStoreRequest(StoreRequest storeRequest) { + if (storeRequest == null){ + flushOnCondition(currFileChannel); + } + if (storeRequest instanceof SyncFlushRequest) { + syncFlush((SyncFlushRequest) storeRequest); + } else if (storeRequest instanceof AsyncFlushRequest) { + async((AsyncFlushRequest) storeRequest); + } else if (storeRequest instanceof CloseFileRequest) { + closeAndFlush((CloseFileRequest) storeRequest); + } + } + + private void closeAndFlush(CloseFileRequest req) { + long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get(); + flush(req.getFileChannel()); + FILE_FLUSH_NUM.addAndGet(diff); + closeFile(currRaf); + } + + private void async(AsyncFlushRequest req) { + if (req.getCurFileTrxNum() < FILE_FLUSH_NUM.get()) { + flushOnCondition(req.getCurFileChannel()); + } + } + + private void syncFlush(SyncFlushRequest req) { + if (req.getCurFileTrxNum() < FILE_FLUSH_NUM.get()) { + long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get(); + flush(req.getCurFileChannel()); + FILE_FLUSH_NUM.addAndGet(diff); + } + // notify + req.wakeupCustomer(); + } + + private void flushOnCondition(FileChannel fileChannel) { + if (FLUSH_DISK_MODE == FlushDiskMode.SYNC_MODEL) { + return; + } + long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get(); + if (diff == 0) { + return; + } + if (diff % MAX_FLUSH_NUM == 0 || + System.currentTimeMillis() - lastModifiedTime > MAX_FLUSH_TIME_MILLS) { + flush(fileChannel); + FILE_FLUSH_NUM.addAndGet(diff); + } + } + + private void flush(FileChannel fileChannel) { + try { + fileChannel.force(false); + } catch (IOException exx) { + LOGGER.error("flush error:" + exx.getMessage()); + } + } + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java new file mode 100755 index 000000000..34f9ba331 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java @@ -0,0 +1,44 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.store; + +/** + * + * @author lizhao + */ +public enum FlushDiskMode { + /** + * sync flush disk + */ + SYNC_MODEL("sync"), + /** + * async flush disk + */ + ASYNC_MODEL("async"); + + private String modeStr; + + FlushDiskMode(String modeStr) { + this.modeStr = modeStr; + } + + public static FlushDiskMode findDiskMode(String modeStr) { + if (SYNC_MODEL.modeStr.equals(modeStr)){ + return SYNC_MODEL; + } + return ASYNC_MODEL; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java new file mode 100755 index 000000000..23b5dce75 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java @@ -0,0 +1,38 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.store; + +/** + * The interface Session storable. + * + * @author jimin.jm @alibaba-inc.com + */ +public interface SessionStorable { + + /** + * Encode byte [ ]. + * + * @return the byte [ ] + */ + byte[] encode(); + + /** + * Decode. + * + * @param src the src + */ + void decode(byte[] src); +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java new file mode 100755 index 000000000..37e2b0f3b --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java @@ -0,0 +1,56 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.store; + +import io.seata.config.Configuration; +import io.seata.config.ConfigurationFactory; + +import static io.seata.core.constants.ConfigurationKeys.STORE_FILE_PREFIX; + + +/** + * @author lizhao + */ +public class StoreConfig { + + private static final Configuration CONFIGURATION = ConfigurationFactory.getInstance(); + + + // default 16kb + private static final int DEFAULT_MAX_BRANCH_SESSION_SIZE = 1024 * 16; + + // default 512b + private static final int DEFAULT_MAX_GLOBAL_SESSION_SIZE = 512; + + // default 16kb + private static final int DEFAULT_WRITE_BUFFER_SIZE = 1024 * 16; + + public static int getMaxBranchSessionSize() { + return CONFIGURATION.getInt(STORE_FILE_PREFIX + "max-branch-session-size", DEFAULT_MAX_BRANCH_SESSION_SIZE); + } + + public static int getMaxGlobalSessionSize() { + return CONFIGURATION.getInt(STORE_FILE_PREFIX + "max-global-session-size", DEFAULT_MAX_GLOBAL_SESSION_SIZE); + } + + public static int getFileWriteBufferCacheSize() { + return CONFIGURATION.getInt(STORE_FILE_PREFIX + "file-write-buffer-cache-size", DEFAULT_WRITE_BUFFER_SIZE); + } + + public static FlushDiskMode getFlushDiskMode() { + return FlushDiskMode.findDiskMode(CONFIGURATION.getConfig(STORE_FILE_PREFIX + "flush-disk-mode")); + } +} \ No newline at end of file diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java new file mode 100755 index 000000000..da7d44a4a --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java @@ -0,0 +1,118 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.store; + +import java.util.List; + +/** + * The interface Transaction store manager. + * + * @author jimin.jm @alibaba-inc.com + */ +public interface TransactionStoreManager { + + /** + * Write session boolean. + * + * @param logOperation the log operation + * @param session the session + * @return the boolean + */ + boolean writeSession(LogOperation logOperation, SessionStorable session); + + /** + * Shutdown. + */ + void shutdown(); + + /** + * Read write store from file list. + * + * @param readSize the read size + * @param isHistory the is history + * @return the list + */ + List readWriteStoreFromFile(int readSize, boolean isHistory); + + /** + * Has remaining boolean. + * + * @param isHistory the is history + * @return the boolean + */ + boolean hasRemaining(boolean isHistory); + + /** + * The enum Log operation. + */ + enum LogOperation { + + /** + * Global add log operation. + */ + GLOBAL_ADD((byte)1), + /** + * Global update log operation. + */ + GLOBAL_UPDATE((byte)2), + /** + * Global remove log operation. + */ + GLOBAL_REMOVE((byte)3), + /** + * Branch add log operation. + */ + BRANCH_ADD((byte)4), + /** + * Branch update log operation. + */ + BRANCH_UPDATE((byte)5), + /** + * Branch remove log operation. + */ + BRANCH_REMOVE((byte)6); + + private byte code; + + LogOperation(byte code) { + this.code = code; + } + + /** + * Gets code. + * + * @return the code + */ + public byte getCode() { + return this.code; + } + + /** + * Gets log operation by code. + * + * @param code the code + * @return the log operation by code + */ + public static LogOperation getLogOperationByCode(byte code) { + for (LogOperation temp : values()) { + if (temp.getCode() == code) { + return temp; + } + } + throw new IllegalArgumentException("Unknown LogOperation[" + code + "]"); + } + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java new file mode 100755 index 000000000..f7ab6c542 --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java @@ -0,0 +1,129 @@ +/* + * Copyright 1999-2019 Seata.io Group. + * + * Licensed 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 io.seata.server.store; + +import java.nio.ByteBuffer; + +import io.seata.common.exception.ShouldNeverHappenException; +import io.seata.server.session.BranchSession; +import io.seata.server.session.GlobalSession; +import io.seata.server.store.TransactionStoreManager.LogOperation; + +/** + * The type Transaction write store. + * + * @author jimin.jm @alibaba-inc.com + * @date 2018 /12/11 + */ +public class TransactionWriteStore implements SessionStorable { + private SessionStorable sessionRequest; + private LogOperation operate; + + /** + * Instantiates a new Transaction write store. + * + * @param sessionRequest the session request + * @param operate the operate + */ + public TransactionWriteStore(SessionStorable sessionRequest, LogOperation operate) { + this.sessionRequest = sessionRequest; + this.operate = operate; + } + + /** + * Instantiates a new Transaction write store. + */ + public TransactionWriteStore() {} + + /** + * Gets session request. + * + * @return the session request + */ + public SessionStorable getSessionRequest() { + return sessionRequest; + } + + /** + * Sets session request. + * + * @param sessionRequest the session request + */ + public void setSessionRequest(SessionStorable sessionRequest) { + this.sessionRequest = sessionRequest; + } + + /** + * Gets operate. + * + * @return the operate + */ + public LogOperation getOperate() { + return operate; + } + + /** + * Sets operate. + * + * @param operate the operate + */ + public void setOperate(LogOperation operate) { + this.operate = operate; + } + + @Override + public byte[] encode() { + byte[] bySessionRequest = this.sessionRequest.encode(); + byte byOpCode = this.getOperate().getCode(); + int len = bySessionRequest.length + 1; + byte[] byResult = new byte[len]; + ByteBuffer byteBuffer = ByteBuffer.wrap(byResult); + byteBuffer.put(bySessionRequest); + byteBuffer.put(byOpCode); + return byResult; + } + + @Override + public void decode(byte[] src) { + ByteBuffer byteBuffer = ByteBuffer.wrap(src); + byte[] bySessionRequest = new byte[src.length - 1]; + byteBuffer.get(bySessionRequest); + byte byOpCode = byteBuffer.get(); + this.operate = LogOperation.getLogOperationByCode(byOpCode); + SessionStorable tmpSessionStorable = getSessionInstanceByOperation(this.operate); + tmpSessionStorable.decode(bySessionRequest); + this.sessionRequest = tmpSessionStorable; + } + + private SessionStorable getSessionInstanceByOperation(LogOperation logOperation) { + SessionStorable sessionStorable = null; + switch (logOperation) { + case GLOBAL_ADD: + case GLOBAL_UPDATE: + case GLOBAL_REMOVE: + sessionStorable = new GlobalSession(); + break; + case BRANCH_ADD: + case BRANCH_UPDATE: + case BRANCH_REMOVE: + sessionStorable = new BranchSession(); + break; + default: + throw new ShouldNeverHappenException("incorrect logOperation"); + } + return sessionStorable; + } +} diff --git a/pigx-visual/pigx-seata-server/src/main/resources/logback.xml b/pigx-visual/pigx-seata-server/src/main/resources/logback.xml new file mode 100755 index 000000000..14508c9dd --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/resources/logback.xml @@ -0,0 +1,54 @@ + + + + + + + + + %date{yyyy-MM-dd HH:mm:ss.SSS} %-5level[%thread]%logger{56}.%method:%L -%msg%n + + + + + ${LOG_HOME}/seata-server.log + true + + ${LOG_HOME}/seata-server.log.%d{yyyy-MM-dd}.%i + 2GB + 7 + 7GB + true + + + %date %level %msg%n%n + UTF-8 + + + + + + + + + + + + + + diff --git a/pigx-visual/pigx-seata-server/src/main/resources/registry.conf b/pigx-visual/pigx-seata-server/src/main/resources/registry.conf new file mode 100755 index 000000000..79b92577a --- /dev/null +++ b/pigx-visual/pigx-seata-server/src/main/resources/registry.conf @@ -0,0 +1,71 @@ +registry { + # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa + type = "nacos" + + nacos { + serverAddr = "localhost" + namespace = "public" + cluster = "default" + } + eureka { + serviceUrl = "http://localhost:1001/eureka" + application = "default" + weight = "1" + } + redis { + serverAddr = "localhost:6379" + db = "0" + } + zk { + cluster = "default" + serverAddr = "127.0.0.1:2181" + session.timeout = 6000 + connect.timeout = 2000 + } + consul { + cluster = "default" + serverAddr = "127.0.0.1:8500" + } + etcd3 { + cluster = "default" + serverAddr = "http://localhost:2379" + } + sofa { + serverAddr = "127.0.0.1:9603" + application = "default" + region = "DEFAULT_ZONE" + datacenter = "DefaultDataCenter" + cluster = "default" + group = "SEATA_GROUP" + addressWaitTime = "3000" + } + file { + name = "file.conf" + } +} + +config { + # file、nacos 、apollo、zk、consul + type = "nacos" + + nacos { + serverAddr = "pigx-nacos" + namespace = "public" + cluster = "default" + } + consul { + serverAddr = "127.0.0.1:8500" + } + apollo { + app.id = "seata-server" + apollo.meta = "http://192.168.1.204:8801" + } + zk { + serverAddr = "127.0.0.1:2181" + session.timeout = 6000 + connect.timeout = 2000 + } + file { + name = "file.conf" + } +} diff --git a/pigx-visual/pom.xml b/pigx-visual/pom.xml index aff244b75..69c5c19aa 100644 --- a/pigx-visual/pom.xml +++ b/pigx-visual/pom.xml @@ -38,6 +38,7 @@ pigx-daemon-quartz pigx-monitor pigx-mp-manager + pigx-seata-server pigx-sso-client-demo pigx-tx-manager