[GitHub] takumiCX edited a comment on issue #352: 修复CompositeOmegaCallback中的compensate方法可能会出现的并发异常

[GitBox <gi...@apache.org>]

takumiCX edited a comment on issue #352: 修复CompositeOmegaCallback中的compensate方法可能会出现的并发异常
URL: https://github.com/apache/servicecomb-pack/pull/352#issuecomment-447717105
 
 
   
   > @takumiCX thanks for your contribute on the servicecomb !
   > 
   > Can you add an unit test to reproduce this issue ? I think this could be very useful for us to understand it. Also would you please to raise an JIRA for this issue on https://issues.apache.org/jira/browse/SCB
   
   ```
   public void compensate(TxEvent event) {
    Map<String, OmegaCallback> serviceCallbacks = callbacks.getOrDefault(event.serviceName(), emptyMap());
   if (serviceCallbacks.isEmpty()) {
    throw new AlphaException("No such omega callback found for service " + event.serviceName());
    }
   OmegaCallback omegaCallback = serviceCallbacks.get(event.instanceId());
    if (omegaCallback == null) {
    LOG.info("Cannot find the service with the instanceId {}, call the other instance.", event.instanceId());
    omegaCallback = serviceCallbacks.values().iterator().next();
    }
   try {
    omegaCallback.compensate(event);
    } catch (Exception e) {
    serviceCallbacks.values().remove(omegaCallback);
    throw e;
    }
    }
   ```
   
   线程可能通过了 if (omegaCallback == null) 的判断条件但是在omegaCallback = serviceCallbacks.values().iterator().next()之前失去了cpu执行权,由于其他线程对serviceCallbacks这个map的操作,目前看alpha端有两种情况:一种是接收到omega端的onDisconnected请求将对应omega端实例从map中移除;一种是执行pendingTask的线程重新进行补偿时失败执行下面这部分代码catch (Exception e)
   
   { serviceCallbacks.values().remove(omegaCallback); throw e; }
   
   时也会移除map中对应的omega端实例。
   
   我觉得我在issue中分析的已经够清晰了。这种并发异常发生的可能性非常小,我是发现了可能存在的bug后通过编写例子验证了我的想法,不过需要在源码上做些改动,这种比较容易复现。
   
   ```
   if (omegaCallback == null) {
    LOG.info("Cannot find the service with the instanceId {}, call the other instance.", event.instanceId());
    try {
    TimeUnit.SECONDS.sleep(2);
    } catch (InterruptedException e) {
    e.printStackTrace();
    }
   
    omegaCallback = serviceCallbacks.values().iterator().next();
   
   }
    
   try {
   
    throw new RuntimeException();
   // omegaCallback.compensate(event);
    } catch (Exception e) {
   ```
   
   然后下面是我写的测试代码
   ```
   @Test
     public void compensateWithConcurrency() throws InterruptedException {
   
       ConcurrentHashMap<String, OmegaCallback> serviceCallbacks = new ConcurrentHashMap();
       serviceCallbacks.put(instanceId1One,callback1One);
       callbacks.put(serviceName1,serviceCallbacks);
       new Thread(new Runnable() {
         @Override
         public void run() {
           compositeOmegaCallback.compensate(eventOf(serviceName1,instanceId1Two,TxStartedEvent));
         }
       }).start();
   
   
       TimeUnit.SECONDS.sleep(1);
   
       new Thread(new Runnable() {
         @Override
         public void run() {
           compositeOmegaCallback.compensate(eventOf(serviceName1,instanceId1One,TxStartedEvent));
         }
       }).start();
   
       TimeUnit.SECONDS.sleep(3);
   
     }
   
   ```

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services