zianxin/pigx
1
0 Fork 0
mirror of http://git.pig4cloud.com/pig/pigx.git synced 2026-11-08 20:49:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

🔖 Releasing / Version tags. 3.1.0

Browse Source
This commit is contained in:
pigxcloud committed 2019-06-05 17:37:05 +08:00
1 parent 178836829d
commit 9aa8b74e5c
33 files changed
-4832
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pigx-visual/pigx-seata-server/pom.xml
-48
View File
@@ -1,48 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>com.pig4cloud</groupId>
<artifactId>pigx-visual</artifactId>
<version>3.1.0</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pigx-seata-server</artifactId>
<packaging>jar</packaging>
<name>pigx-seata-server</name>
<dependencies>
<dependency>
<groupId>io.seata</groupId>
<artifactId>seata-core</artifactId>
<version>0.5.2</version>
</dependency>
<dependency>
<groupId>io.seata</groupId>
<artifactId>seata-config-all</artifactId>
<version>0.5.2</version>
</dependency>
<dependency>
<groupId>io.seata</groupId>
<artifactId>seata-discovery-all</artifactId>
<version>0.5.2</version>
</dependency>
</dependencies>
</project>
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java
-240
View File
@@ -1,240 +0,0 @@
/*
* 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<GlobalBeginRequest, GlobalBeginResponse>() {
@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<GlobalCommitRequest, GlobalCommitResponse>() {
@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<GlobalRollbackRequest, GlobalRollbackResponse>() {
@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<BranchRegisterRequest, BranchRegisterResponse>() {
@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<BranchReportRequest, BranchReportResponse>() {
@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<GlobalLockQueryRequest, GlobalLockQueryResponse>() {
@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<GlobalStatusRequest, GlobalStatusResponse>() {
@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;
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java
-78
View File
@@ -1,78 +0,0 @@
/*
* 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);
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java
-96
View File
@@ -1,96 +0,0 @@
/*
* 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);
}
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java
-55
View File
@@ -1,55 +0,0 @@
/*
* 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;
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java
-46
View File
@@ -1,46 +0,0 @@
/*
* 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;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java
-362
View File
@@ -1,362 +0,0 @@
/*
* 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<GlobalSession> 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<GlobalSession> 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<GlobalSession> 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<GlobalSession> 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();
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java
-310
View File
@@ -1,310 +0,0 @@
/*
* 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();
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java
-186
View File
@@ -1,186 +0,0 @@
/*
* 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<String/* resourceId */,
ConcurrentHashMap<String/* tableName */,
ConcurrentHashMap<Integer/* bucketId */,
Map<String/* pk */, Long/* transactionId */>>>>
LOCK_MAP = new ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>>>();
@Override
public boolean acquireLock(BranchSession branchSession) throws TransactionException {
String resourceId = branchSession.getResourceId();
long transactionId = branchSession.getTransactionId();
ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>> dbLockMap = LOCK_MAP.get(resourceId);
if (dbLockMap == null) {
LOCK_MAP.putIfAbsent(resourceId,
new ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>>());
dbLockMap = LOCK_MAP.get(resourceId);
}
ConcurrentHashMap<Map<String, Long>, Set<String>> 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<Integer, Map<String, Long>> tableLockMap = dbLockMap.get(tableName);
if (tableLockMap == null) {
dbLockMap.putIfAbsent(tableName, new ConcurrentHashMap<Integer, Map<String, Long>>());
tableLockMap = dbLockMap.get(tableName);
}
String[] pks = mergedPKs.split(",");
for (String pk : pks) {
int bucketId = pk.hashCode() % BUCKET_PER_TABLE;
Map<String, Long> bucketLockMap = tableLockMap.get(bucketId);
if (bucketLockMap == null) {
tableLockMap.putIfAbsent(bucketId, new HashMap<String, Long>());
bucketLockMap = tableLockMap.get(bucketId);
}
synchronized (bucketLockMap) {
Long lockingTransactionId = bucketLockMap.get(pk);
if (lockingTransactionId == null) {
// No existing lock
bucketLockMap.put(pk, transactionId);
Set<String> keysInHolder = bucketHolder.get(bucketLockMap);
if (keysInHolder == null) {
bucketHolder.putIfAbsent(bucketLockMap, new ConcurrentSet<String>());
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<String, ConcurrentHashMap<Integer, Map<String, Long>>> 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<Integer, Map<String, Long>> tableLockMap = dbLockMap.get(tableName);
if (tableLockMap == null) {
continue;
}
String[] pks = mergedPKs.split(",");
for (String pk : pks) {
int bucketId = pk.hashCode() % BUCKET_PER_TABLE;
Map<String, Long> 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<Map<String, Long>, Set<String>> lockHolder = branchSession.getLockHolder();
if (lockHolder.size() == 0) {
return true;
}
Iterator<Entry<Map<String, Long>, Set<String>>> it = lockHolder.entrySet().iterator();
while (it.hasNext()) {
Entry<Map<String, Long>, Set<String>> entry = it.next();
Map<String, Long> bucket = entry.getKey();
Set<String> 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;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java
-63
View File
@@ -1,63 +0,0 @@
/*
* 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;
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java
-46
View File
@@ -1,46 +0,0 @@
/*
* 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;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java
-191
View File
@@ -1,191 +0,0 @@
/*
* 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<Long, GlobalSession> 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<GlobalSession> allSessions() {
return sessionMap.values();
}
@Override
public List<GlobalSession> findGlobalSessions(SessionCondition condition) {
List<GlobalSession> 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);
}
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java
-395
View File
@@ -1,395 +0,0 @@
/*
* 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<BranchSession>, SessionStorable {
private static final Logger LOGGER = LoggerFactory.getLogger(BranchSession.class);
private static final int MAX_BRANCH_SESSION_SIZE = StoreConfig.getMaxBranchSessionSize();
private static ThreadLocal<ByteBuffer> 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<Map<String, Long>, Set<String>> lockHolder
= new ConcurrentHashMap<Map<String, Long>, Set<String>>();
/**
* 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<Map<String, Long>, Set<String>> 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());
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java
-65
View File
@@ -1,65 +0,0 @@
/*
* 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<TransactionWriteStore> readWriteStoreFromFile(int readSize, boolean isHistory) {
return null;
}
@Override
public boolean hasRemaining(boolean isHistory) {
return false;
}
};
}
@Override
public void destroy() {
transactionStoreManager.shutdown();
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java
-232
View File
@@ -1,232 +0,0 @@
/*
* 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<Long, BranchSession> 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<Map.Entry<Long, GlobalSession>> 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<Long, BranchSession> unhandledBranchBuffer) {
while (transactionStoreManager.hasRemaining(isHistory)) {
List<TransactionWriteStore> stores = transactionStoreManager.readWriteStoreFromFile(READ_SIZE, isHistory);
restore(stores, unhandledBranchBuffer);
}
}
private void restore(List<TransactionWriteStore> stores, Map<Long, BranchSession> 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();
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java
-541
View File
@@ -1,541 +0,0 @@
/*
* 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<ByteBuffer> 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<BranchSession> 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<SessionLifecycleListener> 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<BranchSession> getSortedBranches() {
ArrayList<BranchSession> sorted = new ArrayList();
sorted.addAll(branchSessions);
return sorted;
}
/**
* Gets reverse sorted branches.
*
* @return the reverse sorted branches
*/
public ArrayList<BranchSession> getReverseSortedBranches() {
ArrayList<BranchSession> 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> T lockAndExcute(LockCallable<T> 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> {
V call() throws TransactionException;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java
-42
View File
@@ -1,42 +0,0 @@
/*
* 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;
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java
-29
View File
@@ -1,29 +0,0 @@
/*
* 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();
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java
-55
View File
@@ -1,55 +0,0 @@
/*
* 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;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java
-121
View File
@@ -1,121 +0,0 @@
/*
* 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;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java
-242
View File
@@ -1,242 +0,0 @@
/*
* 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<GlobalSession> 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<BranchSession> 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();
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java
-89
View File
@@ -1,89 +0,0 @@
/*
* 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;
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java
-90
View File
@@ -1,90 +0,0 @@
/*
* 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;
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java
-109
View File
@@ -1,109 +0,0 @@
/*
* 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<GlobalSession> allSessions();
/**
* Find global sessions list.
*
* @param condition the condition
* @return the list
*/
List<GlobalSession> findGlobalSessions(SessionCondition condition);
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java
-590
View File
@@ -1,590 +0,0 @@
/*
* 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<Runnable>(),
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<GlobalSession> globalSessionsOverMaxTimeout =
sessionManager.findGlobalSessions(new SessionCondition(MAX_TRX_TIMEOUT_MILLS));
if (CollectionUtils.isEmpty(globalSessionsOverMaxTimeout)) {
return true;
}
List<byte[]> 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<BranchSession> 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<TransactionWriteStore> 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<TransactionWriteStore> parseDataFile(File file, int readSize, long currentOffset) {
List<TransactionWriteStore> 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<StoreRequest> 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());
}
}
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java
-44
View File
@@ -1,44 +0,0 @@
/*
* 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;
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java
-38
View File
@@ -1,38 +0,0 @@
/*
* 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);
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java
-56
View File
@@ -1,56 +0,0 @@
/*
* 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"));
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java
-118
View File
@@ -1,118 +0,0 @@
/*
* 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<TransactionWriteStore> 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 + "]");
}
}
}
pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java
-129
View File
@@ -1,129 +0,0 @@
/*
* 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;
}
}
pigx-visual/pigx-seata-server/src/main/resources/logback.xml
-54
View File
@@ -1,54 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<configuration scan="true" scanPeriod="60 seconds" debug="false">
<property name="LOG_HOME" value="${user.home}/logs/seata"/>
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<Pattern>%date{yyyy-MM-dd HH:mm:ss.SSS} %-5level[%thread]%logger{56}.%method:%L -%msg%n</Pattern>
</encoder>
</appender>
<appender name="seata-default"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${LOG_HOME}/seata-server.log</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>${LOG_HOME}/seata-server.log.%d{yyyy-MM-dd}.%i</fileNamePattern>
<maxFileSize>2GB</maxFileSize>
<MaxHistory>7</MaxHistory>
<totalSizeCap>7GB</totalSizeCap>
<cleanHistoryOnStart>true</cleanHistoryOnStart>
</rollingPolicy>
<encoder>
<Pattern>%date %level %msg%n%n</Pattern>
<charset>UTF-8</charset>
</encoder>
</appender>
<logger name="io.seata.server.store.FileTransactionStoreManager" additivity="false">
<level value="INFO"/>
<appender-ref ref="seata-default"/>
</logger>
<root level="DEBUG">
<appender-ref ref="seata-default"/>
<appender-ref ref="stdout"/>
</root>
</configuration>
pigx-visual/pigx-seata-server/src/main/resources/registry.conf
-71
View File
@@ -1,71 +0,0 @@
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"
}
}
pigx-visual/pom.xml
-1
View File
@@ -38,7 +38,6 @@
<module>pigx-daemon-quartz</module>
<module>pigx-monitor</module>
<module>pigx-mp-manager</module>
<module>pigx-seata-server</module>
<module>pigx-sso-client-demo</module>
<module>pigx-tx-manager</module>
</modules>