Java/Spring MVC: provide request context to child threads

I have the Problem, that I want to outsource some processes of my Spring WebMVC application into separate Threads. That was easy enough and works, until I want to use a class, userRightService, which uses the global request. That's not available in the threads, and we get a problem, that's pretty much understandable.

This is my Error:

java.lang.RuntimeException: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.userRightsService': Scope 'request' is not active for the current thread; consider defining a scoped proxy for this bean if you intend to refer to it from a singleton; nested exception is  java.lang.IllegalStateException: Cannot ask for request attribute -  request is not active anymore! 

Okay, clear enough. I am trying to keep the request context by implementing this solution:

How to enable request scope in async task executor

This is my runnable class:

@Scope(value = "request", proxyMode = ScopedProxyMode.TARGET_CLASS) public class myThread implements Runnable {    private RequestAttributes context;    public DataExportThread(RequestAttributes context) {     this.context = context;   }    public void run() {     RequestContextHolder.setRequestAttributes(context); 

And this where it gets spawned:

final DataExportThread dataExportThread =     new myThread(RequestContextHolder.currentRequestAttributes());  final Thread thread = new Thread(myThread); thread.setUncaughtExceptionHandler((t, e) -> {...}); thread.start(); 

As far as I understood, we store the currentRequestAttributes in the thread and then, when running, we restore them currentRequestAttributes... sounded solid to me, but the error is still there. I think I made some mistake adapting the solution for my case. maybe someone can help me finding the error.

Before I went through a lot of stackoverflow-threads with different solutions (see below), so I could try something else next, but this one seemed the clearest and simplest to me, so I hope someone could help me finding the mistake in the implementation or explain why it's the wrong approach.

• Scope 'session' is not active for the current thread; IllegalStateException: No thread-bound request found
• Accessing request scoped beans in a multi-threaded web application
• Using a request scoped bean outside of an actual web request
• InheritableThreadLocal value not inherited by ExecutorService threads
• How to enable request scope in async task executor
• Setting ThreadContext for all threads in application
• Spring mvc request scope context across threads

I already tried this one without success:

• How to enable request scope in async task executor

If it's matters:

<org.springframework-version>4.3.4.RELEASE</org.springframework-version> 

BTW: I know that it would be better to restructure the application in a way, that the request is not needed in the thread but that's very complicated in that case and I really hope I could avoid this.

--

Edit1:

The Bean which can not be created in the thread starts like this:

@Service("userRightsService") @Scope(value = "request", proxyMode = ScopedProxyMode.TARGET_CLASS) public class UserRightsService { 

--

Edit2:

I also tried this one:

• Setting ThreadContext for all threads in application

But context is always empty...

1 Answers

Answers 1

I couldn't reproduce the problem as I am not sure how are you creating/injecting the UserRightsService but I have a couple of suggestions that you may try.

I guess that the problem is that the RequestAttributes is invalidated as the request is over (that's why the exception says Cannot ask for request attribute - request is not active anymore), which happens as your task is running.

Instead, you could try injecting the UserRightsService where your thread is spawned and pass this instance as an argument to the thread. That way the UserRightsService should be created without problem as the request should be still available.

Even so, trying to access the RequestAttributes after the request is over will probably fail. In that case I propose to make a copy of all the values that you need before the request is over, i.e. before your run the thread.

If that doesn't work for you please provide some more info regarding how you initialize the UserRightsService inside the task.

Good luck!

P.S.: I think that the scope annotation in your thread class is useless as the task object is created manually and not managed by spring.

Read More

NullReferenceException in System.Threading.Tasks.RangeWorker.FindNewWork in Android Xamarin app

In my Android native app created with Xamarin, I get the following crash report in HockeyApp:

Xamarin caused by: android.runtime.JavaProxyThrowable: System.NullReferenceException: Object reference not set to an instance of an object   at System.Threading.Tasks.RangeWorker.FindNewWork (System.Int64& nFromInclusiveLocal, System.Int64& nToExclusiveLocal) [0x00000] in <8f1acca5a43d45c5b8d35add5a11806a>:0    at System.Threading.Tasks.RangeWorker.FindNewWork32 (System.Int32& nFromInclusiveLocal32, System.Int32& nToExclusiveLocal32) [0x00000] in <8f1acca5a43d45c5b8d35add5a11806a>:0    at System.Threading._ThreadPoolWaitCallback.PerformWaitCallback () <0xec919968 + 0x00033> in <8f1acca5a43d45c5b8d35add5a11806a>:0 

I cannot figure out what is causing this exception since the stack trace only contains .net code. I have looked at the framework's source code to see if I could make sense of it but to no avail.

I have also tried many different searches in my favorites search engines without finding any posts/articles about similar issues.

From the data associated with those crashes, it seems to be an issue happening exclusively on Samsung devices (S8, S8+ and Note8). I cannot be 100% sure that it doesn't affect other devices but I only have crash reports for those.

Any idea what could cause those crashes? Am I doing something wrong with threading, maybe with cancellation tokens? Are there conditions I am not handling correctly?

Any help to further troubleshoot this issue would be very welcomed.

Thank you

EDIT:

I understand that there is not much to go by but being that this is the only stacktrace I get in HockeyApp and I cannot repro, I have no idea what code causes this.

What I am looking for is more a clue about what could cause Mono threading code to have a null reference exception when I am not managing the threads myself. Or maybe this stacktrace is just a red herring and I need to be looking in other places?

3 Answers

Answers 1

What caught my attention was the FindNewWork32 call -- link for the .NET implementation. Your app may be compiled for 32-bit architecture, and those Samsung devices you mentioned have a 64-bit processor.

This Microsoft paper shows how to target an app to one or more Android-supported CPU architectures. You may need to target multiple platforms:

To target multiple CPU architectures, you can select more than one ABI (at the expense of larger APK file size). You can use the Generate one package (.apk) per selected ABI option (described in Set Packaging Properties) to create a separate APK for each supported architecture.

You do not have to select arm64-v8a or x86_64 to target 64-bit devices; 64-bit support is not required to run your app on 64-bit hardware. For example, 64-bit ARM devices (such as the Nexus 9) can run apps configured for armeabi-v7a. The primary advantage of enabling 64-bit support is to make it possible for your app to address more memory.

Answers 2

I know this answer is not specific for your issue, but when you are dealing with an exception message, which is so brief, it's always a good idea to check system-wide logging. In case of Android, you might get more information using dmesg command (which is giving you the content of /var/log/messages file).

Good luck

Answers 3

You should call .IsCompleted or .Wait() soon after callback. If you do it before the Context objects might go away and if your code is accessing those Context objects will be arbitrarily null (removed). If you call Task.Wait() it would block the thread and then throw an AggregateException once the worker throws it. Try using try catch block to handle exception.

Issue JavaProxyThrowable not always happening due to calling service there might be your activity null some where. You should check null before using activity context.

Read More

Create messaging system in python using socket programming

I am new to socket programming. I wanted to create a simple messaging system between the server and the client ( chat ). I have included my code below. I am expecting it to work as similar as chat system but it doesn't work. If the message is sent it should receive and print it out but only after giving the input the received string is printed. I am expecting it should run parallelly (receive a message and send a message).

Server :

import socket import time import threading  def get(s):     tm = s.recv(1024)     print("\nReceived: ",tm.decode('ascii'))  def set_(s):     i=input("\nEnter : ")     s.send(i.encode('ascii'))   serversocket = socket.socket()  host = socket.gethostname()  port = 9981  serversocket.bind((host,port))  serversocket.listen(1)  clientsocket,addr = serversocket.accept()  while(1):     t1=threading.Thread( target = get ,  args = (clientsocket,) )     t1.start()     t2=threading.Thread( target = set_ ,  args = (clientsocket,) )     t2.start()     time.sleep(10) clientsocket.close() 

Client:

import socket import threading import time def get(s):     tm = s.recv(1024)     print("\nReceived: ",tm.decode('ascii'))      def set_(s):     i=input("\nEnter : ")     s.send(i.encode('ascii'))  s = socket.socket() host = socket.gethostname() port = 9981 s.connect((host,port))  while(1):     t1=threading.Thread( target = get ,  args = (s,) )     t2=threading.Thread( target = set_ , args = (s,) )     t1.start()     t2.start()     time.sleep(10) s.close() 

Output (At Client) :

Enter: hello ------------------------------>(1)  Received: hello --------------------------->(3) 

Output (At Server) :

Enter: hello ------------------------------>(2)  Received :  hello ------------------------->(4) 

Expected Output:

Output (At Client) :

Enter: hello ------------------------------>(1)  Received: hello --------------------------->(4) 

Output (At Server) :

Received :  hello ------------------------->(2)  Enter: hello ------------------------------>(3) 

The number represents the order of execution.

1 Answers

Answers 1

There is an issue with the threading logic of your program. You should move the while(True) loops to the thread workers, and only start your threads once. As it stands, your code can only send/receive one message every 10 seconds.

Server:

import socket import threading  def get(s):     while True:         tm = s.recv(1024)         print("\nReceived: ",tm.decode('ascii'))  def set_(s):     while True:         i=input("\nEnter : ")         s.send(i.encode('ascii'))  serversocket = socket.socket() host = socket.gethostname() port = 9981 serversocket.bind((host,port)) serversocket.listen(1) clientsocket,addr = serversocket.accept() t1=threading.Thread( target = get ,  args = (clientsocket,) ) t1.start() t2=threading.Thread( target = set_ ,  args = (clientsocket,) ) t2.start() 

Client:

import socket import threading  def get(s):     while True:         tm = s.recv(1024)         print("\nReceived: ",tm.decode('ascii'))  def set_(s):     while True:         i=input("\nEnter : ")         s.send(i.encode('ascii'))  s = socket.socket() host = socket.gethostname() port = 9981 s.connect((host,port)) t1=threading.Thread( target = get ,  args = (s,) ) t2=threading.Thread( target = set_ , args = (s,) ) t1.start() t2.start() 

You'll need to handle closing the sockets differently, and the enter/received prints get out of sync after the first message due to the multithreaded nature of the program, but the input is still waiting.

Read More

How does Apache spark handle python multithread issues?

According to python's GIL we cannot use threading in CPU bound processes so my question is how does Apache Spark utilize python in multi-core environment?

1 Answers

Answers 1

Multi-threading python issues are separated from Apache Spark internals. Parallelism on Spark is dealt with inside the JVM.

And the reason is that in the Python driver program, SparkContext uses Py4J to launch a JVM and create a JavaSparkContext.

Py4J is only used on the driver for local communication between the Python and Java SparkContext objects; large data transfers are performed through a different mechanism.

RDD transformations in Python are mapped to transformations on PythonRDD objects in Java. On remote worker machines, PythonRDD objects launch Python sub-processes and communicate with them using pipes, sending the user's code and the data to be processed.

PS: I'm not sure if this actually answers your question completely.

Read More

@Async not working in Spring API rest with Interfaces

I'm working with @Async to stored some data in parallel in the database with hibernate. I need to do that because before saving the information to the database I need to run some task that takes several minutes. So I implemented @Async.

The issue is that @Async seems to not be working. Please find the code below:

WebConfig

@Configuration @EnableAsync @EnableWebMvc public class WebConfig extends WebMvcConfigurerAdapter {  } 

StudentServiceImpl:

@Autowired RunSomeTaskService runSomeTaskService;  @Override Transactional public Response saveWithoutWaiting(StudentBO[] students, String username) throws Exception { ... for (StudentBO student : students) {     ....     Future<Response> response = runSomeTaskService.doTasks(student);     //Finish without waiting for doTasks(). }  @Override     Transactional     public Response saveWithWaiting(StudentBO[] students, String username) throws Exception {     ...     for (StudentBO student : students) {         ....         Future<Response> response = runSomeTaskService.doTasks(student);         //Finish and wait for doTasks().         response.get();     } 

RunSomeTaskService:

public interface RunSomeTaskService{     @Async     public Future<Response> doTasks(Student student); } 

RunSomeTaskServiceImpl:

public class RunSomeTaskServiceImpl extends CommonService implements RunSomeTaskService{  Student student; @Override     public Future<Response> doTasks(Student student) {           Response response = new Response();           this.student = student;           //do Task           return new AsyncResult<Response>(response);        } } 

web.xml

<web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"     xsi:schemaLocation="http://java.sun.com/xml/ns/javaee            http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"     version="3.0">      <display-name>Sample Spring Maven Project</display-name>      <servlet>         <servlet-name>mvc-dispatcher</servlet-name>         <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>         <init-param>             <param-name>contextConfigLocation</param-name>             <param-value>/WEB-INF/spring-config.xml</param-value>         </init-param>         <load-on-startup>1</load-on-startup>         <async-supported>true</async-supported>     </servlet>      <servlet-mapping>         <servlet-name>mvc-dispatcher</servlet-name>         <url-pattern>/</url-pattern>     </servlet-mapping>  <filter>     <filter-name>encodingFilter</filter-name>     <filter-class>             org.springframework.web.filter.CharacterEncodingFilter         </filter-class>     <init-param>       <param-name>encoding</param-name>       <param-value>UTF-8</param-value>     </init-param>   </filter>   <filter-mapping>     <filter-name>encodingFilter</filter-name>     <url-pattern>/*</url-pattern>   </filter-mapping>   <filter>     <filter-name>jwtTokenAuthFilter</filter-name>     <filter-class>org.springframework.web.filter.DelegatingFilterProxy</filter-class>   </filter>   <filter-mapping>     <filter-name>jwtTokenAuthFilter</filter-name>     <url-pattern>/*</url-pattern>   </filter-mapping> </web-app> 

spring.config.xml

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans"     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"     xmlns:util="http://www.springframework.org/schema/util"      xmlns:mvc="http://www.springframework.org/schema/mvc"     xmlns:tx="http://www.springframework.org/schema/tx"     xsi:schemaLocation="http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd   http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd   http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.2.xsd   http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd   http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">      <context:annotation-config  />     <context:component-scan base-package="com.app.controller" />     <tx:annotation-driven transaction-manager="transactionManager"/>     <mvc:annotation-driven />      <bean id="dataSource"         class="org.springframework.jdbc.datasource.DriverManagerDataSource">         ...     </bean>      <bean id="mailSender" class="org.springframework.mail.javamail.JavaMailSenderImpl"> ...     </bean>       <bean id="sessionFactory"         class="org.springframework.orm.hibernate5.LocalSessionFactoryBean">         <property name="dataSource" ref="dataSource" />         <property name="annotatedClasses">             <list>                 <value>//every model generated with Hibernate</value>             </list>         </property>         <property name="hibernateProperties">             <props>                 <prop key="hibernate.dialect">org.hibernate.dialect.MySQL5Dialect</prop>                 <prop key="hibernate.show_sql">true</prop>             </props>         </property>     </bean>      <bean id="transactionManager"         class="org.springframework.orm.hibernate5.HibernateTransactionManager">         <property name="sessionFactory" ref="sessionFactory" />     </bean>      <bean id="persistenceExceptionTranslationPostProcessor"         class="org.springframework.dao.annotation.PersistenceExceptionTranslationPostProcessor" />      <bean id="studentService" class="com.app.services.StudentServiceImpl"></bean>     <bean id="studentDao" class="com.app.dao.StudentDaoImpl"></bean>     ...      <bean id="jwtTokenAuthFilter" class="com.app.security.JWTTokenAuthFilter" />        </beans> 

So, could you please help me to understand why @Async is not working?

5 Answers

Answers 1

Here you find the solutions

// servlet.setAsyncSupported(true);

//For Example

public class WebAppInitializer implements WebApplicationInitializer {     @Override     public void onStartup(ServletContext servletContext) throws ServletException {         AnnotationConfigWebApplicationContext ctx = new AnnotationConfigWebApplicationContext();         ctx.register(WebConfig.class);         ctx.setServletContext(servletContext);         ServletRegistration.Dynamic servlet = servletContext.addServlet("dispatcher",             new DispatcherServlet(ctx));         servlet.setLoadOnStartup(1);         servlet.addMapping("/");         servlet.setAsyncSupported(true); //Servlets were marked as supporting async         // For CORS Pre Filght Request         servlet.setInitParameter("dispatchOptionsRequest", "true");     } } 

Answers 2

Well, finally I make it work...

I used Executors in the following way:

ExecutorService executor = Executors.newFixedThreadPool(students.size()); for (StudentBO student : students) {     executor.submit(() -> extractDataService.doTask(student)); } 

Where doTask is a regular function, that when I don't need it to work in a different thread, I just call it as it is. When I need the threads, I use the code above.

Answers 3

More Sophisticated way would be to implement AsyncConfigurer and set the AsyncExecutor to threadPoolTaskExecutor.

Sample Code below

@Configuration @EnableAsync(proxyTargetClass=true) //detects @Async annotation public class AsyncConfig implements AsyncConfigurer {   public Executor threadPoolTaskExecutor() {         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();         executor.setCorePoolSize(10); // create 10 Threads at the time of initialization         executor.setQueueCapacity(10); // queue capacity         executor.setMaxPoolSize(25); // if queue is full, then it will create new thread and go till 25         executor.setThreadNamePrefix("DEMO-");         executor.initialize();//Set up the ExecutorService.         return executor;     }      @Override     public Executor getAsyncExecutor() {         return threadPoolTaskExecutor();     }      @Override     public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {         return new YOUR_CUSTOM_EXCEPTION_HANDLER();     }  }  

The above configuration will detect @Async annotation wherever mentioned

Answers 4

You can do CompletableFuture , with this you know when all your tasks are complete

List<CompletableFuture<T>> futureList = new ArrayList<>();  for(Student student:studentList){  CompletableFuture<T> returnedFuture = CompletableFuture.supplyAsync(() -> doSomething(student),executor).exceptionally(e -> {         log.error("Error occured in print something future",e);         return 0;     });  futureList.add(returnedFuture); }  Completable.allOf(futureList); 

Then you can pipeline with thenCompose or thenApply (to take consumer) to have complete control on the task pipeline. you can shutdonw executors when you are done safely.

CompletetableFuture.allOff javadoc for more info

Answers 5

There is possibility that the @EnableAsync annotation in WebConfig.java is never scanned. The web.xml points to the spring-context.xml.

You can change the DispatcherServlet definition in web.xml to:

<servlet>     <servlet-name>mvc-dispatcher</servlet-name>     <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>     <init-param>         <param-name>contextClass</param-name>         <param-value>             org.springframework.web.context.support.AnnotationConfigWebApplicationContext         </param-value>     </init-param>     <init-param>         <param-name>contextConfigLocation</param-name>         <param-value>             com.yourpackage.WebConfig         </param-value>     </init-param>     <load-on-startup>1</load-on-startup>     <async-supported>true</async-supported> </servlet> 

And include all configuration from spring-config.xml to this class.

Or Add <task:annotation-driven> in spring-config.xml.

Updated

Currently, com.app.controller package is scanned in spring-config.xml. Make sure the WebConfig.java is in this package or one of it's sub-package. If not add WebConfig's package to base package attribute separated by comma.

Additionally, you can control the thread pool used by async task. Create a executor bean

@Bean public Executor asyncTaskExecutor() {     ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();     executor.setCorePoolSize(5);     executor.setMaxPoolSize(10);     executor.setThreadNamePrefix("asynctaskpool-");     executor.initialize();     return executor; }  

And in your async method use the bean name like this

@Async("asyncTaskExecutor") public Future<Response> doTasks(Student student); 

This will ensure all task will be executed in this thread pool.

Read More

Is it possible for a thread that is not the UI thread to manipulate the UI elements?

I have read that only the UI thread should be allowed to manipulate the UI elements in WinAPI. But I don't think that it is even possible for a thread that is not the UI thread to manipulate the UI elements.

I think that because when a thread (that is not the UI thread) calls the SendMessage() function to manipulate some UI element, a message will be sent to the UI thread, and then it is the UI thread that will manipulate the UI element and not the other thread.

Am I correct?

1 Answers

Answers 1

First, hypothetically speaking in an attempt to satisfy the OP's curiosity:

• If we define manipulating UI elements as reading from or writing to elements' properties, then technically you could come up with your own UI framework that would maintain the elements independently from the Windows API. Such attempts have been made. WPF is one of them. You could then theoretically make the framework thread-safe and make it possible to access the elements' properties from multiple threads.
• Also, GDI allows access to its objects from multiple threads, so you could potentially draw to your window from multiple threads (ditto for DirectX). WPF for example has a dedicated render thread (or at least it used to). You could also specify a different thread to process input with AttachThreadInput.

However, given the premise of the question that we're sticking to using the standard Windows API for creating and managing the UI, it is safe to say that access to the window is only achieved from within the thread that created it, because SendMessage() will switch to the owner thread. But that's not to say that invoking SendMessage() from multiple threads is a safe or a recommended approach. On the contrary, it is fraught with peril and care would have to be taken to properly synchronize the threads.

For one thing, a typical WndProc() looks like this:

LRESULT CALLBACK WndProc(HWND hWnd, UINT message, WPARAM wParam, LPARAM lParam) {     ...     switch (message)     {         case WM_MYMSG1:             ...             SendMessage(hWnd, WM_MYMSG2, wParam, lParam);             ...         break;         ...         }     ... } 

So in order to protect your WndProc() so it can be accessed from multiple threads, you would have to make sure to use a reentrant lock, and not a semaphore, for example.

Secondly, if you use a reentrant lock you must make sure that it is only used within WndProc() or even make it specific to a message. Otherwise it is very easy to get into a deadlock:

//Worker thread: void foo ()  {     EnterCriticalSection(&g_cs);     SendMessage(hWnd, WM_MYMSG1, NULL, NULL);     LeaveCriticalSection(&g_cs);  }   //Owner thread: LRESULT CALLBACK WndProc(HWND hWnd, UINT message, WPARAM wParam, LPARAM lParam) {     switch (message)     {         case WM_MYMSG1:         {             EnterCriticalSection(&g_cs); //Deadlock!             ...             LeaveCriticalSection(&g_cs);          }         break;     } } 

Thirdly, you would have to make sure not to invoke any control-yielding functions within your WndProc(); these include but are not limited to: DialogBox(), MessageBox() and GetMessage(). Else you get a deadlock.

Then, consider a multi-window application, with each window's message pump being run in a separate thread. You would have to ensure not to send any messages between the threads in order not to end up with a deadlock:

LRESULT CALLBACK WndProc(HWND hWnd, UINT message, WPARAM wParam, LPARAM lParam) {     ...     switch (message)     {         case WM_MYMSG1:             ...             SendMessage(hWnd2, WM_MYMSG1, wParam, lParam); //Deadlock!             ...         break;         ...         }     ... } 

You would also have to be very careful with using Windows APIs that implicitly manage the operating system's process-specific locks, and preserve and maintain the proper lock hierarchy. Quite a few User32 functions and many blocking COM calls fall into this category.

Some of these issues may be alleviated by using InSendMessage() and ReplyMessage() (when using SendMessage()) or PostMessage() and its siblings. However then you get into all kinds of control flow issues, because you may want to know that the message was processed before continuing the current thread or processing the next message. So you end up having to implement some kind of a synchronization mechanism anyway, but this becomes increasingly difficult with many pitfalls to avoid.

Problems don't stop with just sending messages between threads either. Changing WndProc() from a different thread can lead to terrible race-condition bugs:

//in UI thread: wpOld = (WNDPROC)GetWindowLongPtr(hwnd, GWLP_WNDPROC); //in another thread: SetWindowLongPtr(hwnd, GWLP_WNDPROC, (LONG_PTR)otherWndProc); //back in UI thread: SetWindowLongPtr(hwnd, GWLP_WNDPROC, (LONG_PTR)newWndProc); //still in UI thread: LRESULT CALLBACK newWndProc(...) {     CallWindowProc(wpOld, ...); //Wrong wpOld! } 

Also, improperly using DCs from multiple threads can lead to subtle bugs.

These reasons, and others (including performance), may have led the designers of standard API wrappers like MFC and WinForms to simply assume that their APIs will be used in a single-thread context. They don't offer any thread-safety protections and it's up to the user to implement such mechanisms, however the higher level of abstraction makes it even easier to neglect the underlying issues. When such problems arise, usually the answer is: don't use the control from outside the owner thread.

Read More

Android USB Accessory Multi Thread

Headache caused by multi-threading and Android Open Accessory.

I need to communicate with a USB Accessory, but I need to do it from 2 threads. One thread generates and sends data the other one reads data.

• Why I don't use a single thread? Because there can be 1 or more writes before a read and reads are blocking, so that is not an option.

• If using multiple threads, I do run into I/O Error (No such device) sooner or later, because I will have a collision between read & write being executed at the same time.

• Locking will more or less put me back in single-thread situation, so not good.

• .available() method on the input-stream returns is not supported, so I cannot check if anything is available before doing a read

• Since it's not a socket-based stream I cannot set timeout either.

• I have tried getting the FileDescriptor from the USBAccessory and passing to JNI to handle it there, but after the first read/write the device becomes inaccessible.

Question/Suggestion needed:
What will be a suggested/best-practice approach to this? I do not expect written code, I just need some guidance on how to approach this problem.

To clarify:
The software at the other end might or might NOT respond with any data. There are some so called silent sends were the data sent it's just received but there is no ACK. Since the app I'm working on is only a proxy, I do not have a clear picture if the data will or will not produce an answer. That will require analysis of the data as well, which isn't on the books at the moment.

Thank you.

1 Answers

Answers 1

As you want to do read and write in parallel, writing will always lead to a pause to read if the read is on the same part as write.

May be you can follow similar approach as ConcurrentHashMap and use different locks for different segments and lock read only if write is on the same segment else allow the read to happen.

This will

1. Avoid blocking read during write in most scenarios
2. Avoid collision and
3. Definitely wont be a single thread approach.

Hope that helps.

Read More

moveToThread vs deriving from QThread in Qt

When should moveToThread be preferred over subclassing QThread?

This link shows that both methods work. On what basis should I decide what to use from those two?

5 Answers

Answers 1

I would focus on the differences between the two methods. There isn't a general answer that fits all use cases, so it's good to understand exactly what they are to choose the best that fits your case.

Using moveToThread()

moveToThread() is used to control the object's thread affinity, which basically means setting the thread (or better the Qt event loop) from which the object will emit signals and its slots will be executed.

As shown in the documentation you linked, this can be used to run code on a different thread, basically creating a dummy worker, writing the code to run in a public slot (in the example the doWork() slot) and then using moveToThread to move it to a different event loop.

Then, a signal connected to that slot is fired. Since the object that emits the singal (the Controller in the example) lives in a different thread, and the signal is connected to our doWork method with a queued connection, the doWork method will be executed in the worker thread.

The key here is that you are creating a new event loop, run by the worker thread. Hence, once the doWork slot has started, the whole event loop will be busy until it exits, and this means that incoming signals will be queued.

Subclassing QThread()

The other method described in Qt's documentation is subclassing QThread. In this case, one overrides the default implementation of the QThread::run() method, which creates an event loop, to run something else.

There's nothing wrong with this approach itself, although there are several catches.

First of all, it is very easy to write unsafe code, because the run() method is the only one in that class that will be actually run on another thread.

If as an example, you have a member variable that you initialize in the constructor and then use in the run() method, your member is initialized in the thread of the caller and then used in the new thread.

Same story for any public method that could be called either from the caller or inside run().

Also slots would be executed from the caller's thread, (unless you do something really weird as moveToThread(this)) leading to extra confusion.

So, it is possible, but you really are on your own with this approach and you must pay extra attention.

Other approaches

There are of course alternatives to both approaches, depending on what you need. If you just need to run some code in background while your GUI thread is running you may consider using QtConcurrent::run().

However, keep in mind that QtConcurrent will use the global QThreadPool. If the whole pool is busy (meaning there aren't available threads in the pool), your code will not run immediately.

Another alternative, if you are at the least on C++11, is to use a lower level API such as std::thread.

Answers 2

As a starting point: use neither. In most cases, you have a unit of work that you wish to run asynchronously. Use QtConcurrent::run for that.

If you have an object that reacts to events and/or uses timers, it's a QObject that should be non-blocking and go in a thread, perhaps shared with other objects.

Such an object can also wrap blocking APIs.

Subclassing QThread is never necessary in practice. It's like subclassing QFile. QThread is a thread handle. It wraps a system resource. Overloading it is a bit silly.

Answers 3

QThread is low level thread abstraction, first look at high level API QtConcurrent module and QRunnable

If nothing of these is suitable for you, then read this old article, it tells how you should use QThread. Think about thread and task performed in this thread as a separate objects, don't mix them together.

So, if you need to write come custom, specific or extended thread wrapper then you should subclass QThread.

If you have QObject derived class with signals and slots, then use moveToThread on it.

In other cases use QtConcurrent, QRunnable and QThreadPoll.

Answers 4

Simple answer is ALWAYS. When you move object to thread:

• it is easy to write test for code
• it is easy to refactor code (you can use thread but you don't have to).
• you do not mix functionality of thread with business logic
• there is no problem with object lifetime

When you subclass QThread

• it is harder to write test
• object clean up process can get very confusing leading to strange errors.

There is full description of the problem from Qt blog: You’re doing it wrong….

QtConcurrent::run is also very handy.

Please remember that by default slots are trying to jump between treads when signal is send from other thread object is assigned to. For details see documentation of Qt::ConnectionType.

Answers 5

Here are official guidelines regarding all threads technologies in Qt: http://doc.qt.io/qt-5/threads-technologies.html

Read More

std::locale/std::facet Critical section

Out of curiosity. In the past I've seen performance degradation in function like boost::to_lower because of the CriticalSection employed in std::use_facet when the lazy facet is allocated. As far as I remember there was a bug with global lock on locale but according to Stephan Lavavej it was fixed in VS2013. And voila, I saw this lock on facet killing server performance yesterday so I guess I'm confusing two different issues.
But in the first place, why there is a CriticalSection around the lazy facet? Obviously it will ruin the performance. Why they didnt resolve to some kind of upgradable lock or atomic operations on pointers?

1 Answers

Answers 1

MSVC++'s std::locale is implemented in terms of the underlying C function setlocale. That touches global state, and must therefore be protected by a lock.

Changing the locking semantics of a data structure is unfortunately an ABI breaking change, so not much we'll be able to do about it for a while.

Read More

What could cause a Java ScheduleService to not run?

In my Java application I define a ScheduleService like this:

ScheduledService<Void> scheduledService = new ScheduledService<Void>() {     @Override     protected Task<Void> createTask() {         return new Task<Void>() {             @Override             protected Void call() {                tick();                return null;             }         };     }  }; scheduledService.setPeriod(new javafx.util.Duration(TICK_PERIOD.toMillis())); scheduledService.start(); 

When I trigger the application from IntelliJ it works fine and tick() runs every second. When the application is packaged as an .exe using the JavaFX Packager, the service is never started.

The state of the server after I run .start() in all cases is SCHEDULED. Any ideas what else might be going on? Could something be preventing the creation of threads? Or maybe it's not switching between various threads?

The documentation for ScheduledService says (emphasis mine):

Timing for this class is not absolutely reliable. A very busy event thread might introduce some timing lag into the beginning of the execution of the background Task, so very small values for the period or delay are likely to be inaccurate. A delay or period in the hundreds of milliseconds or larger should be fairly reliable.

Is it possible there's some issue with the event thread? Is there a way to inspect it?

After calling start(), scheduleService.getExecutor() returns null. Is that expected?

I tried setting my own executor defined this way:

BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>(); ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(32, Integer.MAX_VALUE, 1000, TimeUnit.MILLISECONDS, blockingQueue); scheduledService.setExecutor(threadPoolExecutor); 

and then I print it out before and after calling start. Before it looks like this:

java.util.concurrent.ThreadPoolExecutor@4d97d155[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0] 

and afterwards:

java.util.concurrent.ThreadPoolExecutor@4d97d155[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0] 

So, it is claiming there's an active thread, even though it doesn't seem to be active at all.

Update: I removed the mention of a screensaver because I manage to reproduce the issue as a simple .exe but I still have the problem that the problem doesn't happen when running it from IntelliJ, it only happens when packaged as an .exe.

1 Answers

Answers 1

I found the solution and it had nothing to do with ScheduleService. There were literally three other bugs in my app that were compounding to produce the unexpected behavior as well as hide my attempts at exploring the problem.

Read More

The relationship between thread and process in multi-process program

OS: debian9.
A simple multi-processes program named mprocesses.py.

import os import multiprocessing  def run_task(name):     print("task %s (pid = %s) is running"  %(name,os.getpid()))     while True:         pass  if __name__ == "__main__":     print("current process %s ." %os.getpid())     pool = multiprocessing.Pool(processes = 2)     for i in range(2):         pool.apply_async(run_task,args=(i,))     pool.close()     pool.join() 

Run python3 mprocesses.py and get below output.

python3 mprocesses.py current process 6145 . task 0 (pid = 6146) is running task 1 (pid = 6147) is running 

Get processes info.

ps lax |grep 'python3 mprocesses.py' |grep -v grep  0  1000  6145  5615  20   0 275428 14600 -      Sl+  pts/1      0:00 python3 mprocesses.py 1  1000  6146  6145  20   0  54232 10340 -      R+   pts/1      1:01 python3 mprocesses.py 1  1000  6147  6145  20   0  54232 10348 -      R+   pts/1      1:01 python3 mprocesses.py 

Check processes tree view.

pstree -p 5615 bash(5615)───python3(6145)─┬─python3(6146)                            ├─python3(6147)                            ├─{python3}(6148)                            ├─{python3}(6149)                            └─{python3}(6150) 

What confused me is the three threads 6148,6149,6150.
Does that mean every process contain one process ? Maybe my logical graph is better to express relationships between processes and threads here.

bash(5615)───python3(6145)─┬─────────────────python3(6146)                            |                    └─{python3}(6149)                            |                                         ├──────────────────python3(6147)                            ├─{python3}(6148)     └─{python3}(6150) 

1.bash(5615) is the python3 mprocesses.py(6145) 's father process.
2.python3 mprocesses.py(6145) contains two processes 6146 and 6147 created by pool = multiprocessing.Pool(processes = 2).
3.Process(6145) contain thread(6148),Process(6146) contain thread(6149),Process(6147) contain thread(6150).
It does no matter which exact process id contain which thread id.
Is my understanding right?

1 Answers

Answers 1

You have:

• 3 processes (1 parent process and 2 children to match your processes = 2 argument)
• 2 threads in each process (1 main thread, and 1 communication and management thread)

The extra communication and management thread per process is an implementation detail of the multiprocessing module; if you are sharing resources between processes more threads may be used. You can see hints that threads are used for these tasks in the documentation

For example, under Pipes and Queues:

Note: When an object is put on a queue, the object is pickled and a background thread later flushes the pickled data to an underlying pipe.

[...]

class multiprocessing.Queue([maxsize])
Returns a process shared queue implemented using a pipe and a few locks/semaphores. When a process first puts an item on the queue a feeder thread is started which transfers objects from a buffer into the pipe.

(italic emphasis mine)

You don't need to worry about these threads; they are there to implement the multiprocessing functionality and make it all run smoothly.

Read More

Mysql lock timeout when inserting concurrently

I am trying to insert concurrently (heavy inserts by 8 threads) into sql throught hibernate. My pojo consists of two tables, one table references the other through foreign key constraint. I am trying to save a lot of instances of my pojo to the db concurrently. Sometimes the insert is failing and rolling back because of lock wait timeout.

Caused by: java.sql.SQLException: Lock wait timeout exceeded; try restarting transaction.

Suppose there are table A(table edges in the screenshot) and table B. Table B has a foreign key constraint which references it's id to primary key of table A. What I could infer from the locks table is an S lock is being held on table A's record by trx 114888 while try to insert in table B(id corresponding to table B) and 11493 is waiting to acquire X lock to insert new record in table A. Table A has index on some of its columns.

What is meant by supremum pseudo-record here? Is it a gap lock? If so then why is the record type as 'RECORD'? Is there a way around this so as to avoid this gap lock or whatever lock it is?

These are the screenshots of the lock tables. INNODB_LOCKS table screnshot

Some innodb status logs

---TRANSACTION 114893, ACTIVE 2611 sec inserting mysql tables in use 1, locked 1 LOCK WAIT 48609 lock struct(s), heap size 4726992, 585460 row lock(s), undo log entries 1132742 MySQL thread id 12620, OS thread handle 123145553027072, query id 38123782 localhost 127.0.0.1 root update insert into edges (---some values--) Trx read view will not see trx with id >= 114862, sees < 114817 ------- TRX HAS BEEN WAITING 22 SEC FOR THIS LOCK TO BE GRANTED: RECORD LOCKS space id 408 page no 135298 n bits 240 index PRIMARY of table `**database**.edges` trx id 114893 lock_mode X insert intention waiting Record lock, heap no 1 

1 Answers

Answers 1

You are using the repeatable read isolation level. In the repeatable read isolation level so called gap locks are used and are held for the duration of the transaction (you can read more about gap locks in the documentation). If you switch the isolation level from repeatable read to read committed, the problem will go away. You can set the isolation level with

set transaction isolation level read committed 

You should check the documentation of the command. The isolation level can be set at a session level or global level.

Read More

Separate computation from socket work in Python

I'm serializing column data and then sending it over a socket connection. Something like:

import array, struct, socket  ## Socket setup s = socket.create_connection((ip, addr))  ## Data container setup ordered_col_list = ('col1', 'col2') columns = dict.fromkeys(ordered_col_list)  for i in range(num_of_chunks):     ## Binarize data     columns['col1'] = array.array('i', range(10000))     columns['col2'] = array.array('f', [float(num) for num in range(10000)])     .     .     .      ## Send away     chunk = b''.join(columns[col_name] for col_name in ordered_col_list]     s.sendall(chunk)     s.recv(1000)      #get confirmation 

I wish to separate the computation from the sending, put them on separate threads or processes, so I can keep doing computations while data is sent away.

I've put the binarizing part as a generator function, then sent the generator to a separate thread, which then yielded binary chunks via a queue.

I collected the data from the main thread and sent it away. Something like:

import array, struct, socket from time import sleep try:     import  thread     from Queue import Queue except:     import _thread as thread     from queue import Queue   ## Socket and queue setup s = socket.create_connection((ip, addr)) chunk_queue = Queue()   def binarize(num_of_chunks):     ''' Generator function that yields chunks of binary data. In reality it wouldn't be the same data'''      ordered_col_list = ('col1', 'col2')     columns = dict.fromkeys(ordered_col_list)      for i in range(num_of_chunks):         columns['col1'] = array.array('i', range(10000)).tostring()         columns['col2'] = array.array('f', [float(num) for num in range(10000)]).tostring()         .         .          yield b''.join((columns[col_name] for col_name in ordered_col_list))   def chunk_yielder(queue):     ''' Generate binary chunks and put them on a queue. To be used from a thread '''      while True:            try:             data_gen = queue.get_nowait()         except:             sleep(0.1)             continue         else:                 for chunk in data_gen:                 queue.put(chunk)   ## Setup thread and data generator thread.start_new_thread(chunk_yielder, (chunk_queue,)) num_of_chunks = 100 data_gen = binarize(num_of_chunks) queue.put(data_gen)   ## Get data back and send away while True:    try:         binary_chunk = queue.get_nowait()     except:         sleep(0.1)         continue     else:             socket.sendall(binary_chunk)         socket.recv(1000) #Get confirmation 

However, I did not see and performance imporovement - it did not work faster.

I don't understand threads/processes too well, and my question is whether it is possible (at all and in Python) to gain from this type of separation, and what would be a good way to go about it, either with threads or processess (or any other way - async etc).

2 Answers

Answers 1

If you are trying to use concurrency to improve performance in CPython I would strongly recommend using multiprocessing library instead of multithreading. It is because of GIL (Global Interpreter Lock), which can have a huge impact on execution speed (in some cases, it may cause your code to run slower than single threaded version). Also, if you would like to learn more about this topic, I recommend reading this presentation by David Beazley. Multiprocessing bypasses this problem by spawning a new Python interpreter instance for each process, thus allowing you to take full advantage of multi core architecture.

Answers 2

You have two options for running things in parallel in Python, either use the multiprocessing (docs) library , or write the parallel code in cython and release the GIL. The latter is significantly more work and less applicable generally speaking.

Python threads are limited by the Global Interpreter Lock (GIL), I won't go into detail here as you will find more than enough information online on it. In short, the GIL, as the name suggests, is a global lock within the CPython interpreter that ensures multiple threads do not modify objects, that are within the confines of said interpreter, simultaneously. This is why, for instance, cython programs can run code in parallel because they can exist outside the GIL.

---

As to your code, one problem is that you're running both the number crunching (binarize) and the socket.send inside the GIL, this will run them strictly serially. The queue is also connected very strangely, and there is a NameError but let's leave those aside.

With the caveats already pointed out by Jeremy Friesner in mind, I suggest you re-structure the code in the following manner: you have two processes (not threads) one for binarising the data and the other for sending data. In addition to those, there is also the parent process that started both children, and a queue connecting child 1 to child 2.

• Subprocess-1 does number crunching and produces crunched data into a queue
• Subprocess-2 consumes data from a queue and does socket.send

in code the setup would look something like

from multiprocessing import Process, Queue  work_queue = Queue() p1 = Process(target=binarize, args=(100, work_queue)) p2 = Process(target=send_data, args=(ip, port, work_queue)) p1.start() p2.start() p1.join() p2.join() 

binarize can remain as it is in your code, with the exception that instead of a yield at the end, you add elements into the queue

def binarize(num_of_chunks, q):     ''' Generator function that yields chunks of binary data. In reality it wouldn't be the same data'''      ordered_col_list = ('col1', 'col2')     columns = dict.fromkeys(ordered_col_list)     for i in range(num_of_chunks):         columns['col1'] = array.array('i', range(10000)).tostring()         columns['col2'] = array.array('f', [float(num) for num in range(10000)]).tostring()         data = b''.join((columns[col_name] for col_name in ordered_col_list))         q.put(data) 

send_data should just be the while loop from the bottom of your code, with the connection open/close functionality

def send_data(ip, addr, q):      s = socket.create_connection((ip, addr))      while True:          try:              binary_chunk = q.get(False)          except:              sleep(0.1)              continue          else:                  socket.sendall(binary_chunk)              socket.recv(1000) # Get confirmation     # maybe remember to close the socket before killing the process 

Now you have two (three actually if you count the parent) processes that are processing data independently. You can force the two processes to synchronise their operations by setting the max_size of the queue to a single element. The operation of these two separate processes is also easy to monitor from the process manager on your computer top (Linux), Activity Monitor (OsX), don't remember what it's called under Windows.

---

Finally, Python 3 comes with the option of using co-routines which are neither processes nor threads, but something else entirely. Co-routines are pretty cool from a CS point of view, but a bit of a head scratcher at first. There is plenty of resources to learn from though, like this post on Medium and this talk by David Beazley.

---

Even more generally, you might want to look into the producer/consumer pattern, if you are not already familiar with it.

Read More

Execute SoapUI test on multi-threads

I have a SoapUI test which uses an input file to read lines as input of requests. So there is a loop which reads data and execute request and write output to file. Response times are too long, so processing of this file should be done asynchronously, but I am not sure, how SoapUI can handle this. There is file attachment in SOAP requests, which is not handled by current version of JMeter.

2 Answers

Answers 1

I understood this question as requiring the ability to call a service asynchronously due to the time it takes to process. So, by this, I mean SoapUI makes a request to a web service and instead of waiting for it, it carries on. At some point later, SoapUI receives the response.

SoapUI can handle this, I haven't tried it myself, but when reading some guides recently, I noticed it can be done.

See.... Blog Guide

SoapUI Forum

In short, it involves setting up a mock service to receive the response, which can then be validated.

Answers 2

As per the SoapUI's documentation below, both test cases or test suites can be executed in Parallel mode.

In the case of TestSuites and TestCases these can be executed either in sequence or parallell, as configured with the corresponding toolbar buttons.

In the above image, first one in the marked image stands for sequential execution and the second one (with multiple parallel arrows) stands for Parallel execution mode.

User can select either of the one before executing the tests.

Hope this helps.

Note that SOAPUI does not allows test steps to be executed in parallel. If you need any custom execution i.e., same test case and steps to be executed in Parallel, here is sample project done for that. It can be used as reference and apply it to your case.

Read More

A3C in Tensorflow - Should I use threading or the distributed Tensorflow API

I want to implement the Asynchronous Advantage Actor Critic (A3C) model for reinforcement learning in my local machine (1 CPU, 1 cuda compatible GPU). In this algorithm, several "learner" networks interact with copies of an environment and update a central model periodically.

I've seen implementations that create n "worker" networks and one "global" network inside the same graph and use threading to run these. In these approaches, the global net is updated by applying gradients to the trainable parameters with a "global" scope.

However, I recently read a bit about distributed tensorflow and now I'm a bit confused. Would it be easier/faster/better to implement this using the distributed tensorflow API? In the documentation and talks they always make expicit mention of using it in multi-device environments. I don't know if it's an overkill to use it in a local async algorithm.

I would also like to ask, is there a way to batch the gradients calculated by every worker to be applied together after n steps?

1 Answers

Answers 1

I found using threading simpler than the distributed tensorflow API, however it also runs slower. The more CPU cores you use, the faster distributed tensorflow becomes compared to threads.

However this only holds for asynchronous training. If the available CPU cores are limited and you want to make use of a GPU, you might want to use synchronous training with multiple workers instead, like OpenAI does in their A2C implementation. There only the environments are parallelized (through multiprocessing) and tensorflow uses the GPU without any graph parallelization. OpenAI reported that their results were better with synchronous training than with A3C.

Read More

How do I update string value from inside a thread

I am coding a Xamarin.Forms cross platform app which works with users accounts. The problem is I get their username from my Database but it doesn't ever update the value of public static string username = "";

I am assuming it is because it's being ran inside a Thread or something to do with the WebRequest, I have done research for quiet a while but haven't been able to find a solution.

The method I am using to update their username is as follows

private void loadUserData()     {         username = "Test";         Uri uri = new Uri("http://example.com/session-data.php?session_id=" + session);         WebRequest request = WebRequest.Create(uri);         request.BeginGetResponse((result) =>         {             try             {                 Stream stream = request.EndGetResponse(result).GetResponseStream();                 StreamReader reader = new StreamReader(stream);                 Device.BeginInvokeOnMainThread(() =>                 {                     string page_result = reader.ReadToEnd();                     var jsonReader = new JsonTextReader(new StringReader(page_result))                     {                         SupportMultipleContent = true // This is important!                     };                     var jsonSerializer = new JsonSerializer();                     try                     {                         while (jsonReader.Read())                         {                             UserData userData = jsonSerializer.Deserialize<UserData>(jsonReader);                             username = userData.username;                         }                      }                     catch (Newtonsoft.Json.JsonReaderException readerExp)                     {                         string rEx = readerExp.Message;                         Debug.WriteLine(rEx);                     }                 });             }             catch (Exception exc)             {                 string ex = exc.Message;                 Debug.WriteLine(ex);             }          }, null);     } 

When the url is opened it prints out the following line

{"id":7,"username":"TestUser","name":"Test User","bio":"Hello World","private":0}

UserData contains the following code

class UserData {     [JsonProperty("id")]     public int id { get; set; }      [JsonProperty("username")]     public string username { get; set; }      [JsonProperty("name")]     public string name { get; set; }      [JsonProperty("bio")]     public string bio { get; set; }      [JsonProperty("private")]     public int isPrivate { get; set; } } 

I also noticed the following error prints out, I tried googling around and haven't found any solutions I understand to fix this

Error parsing positive infinity value. Path '', line 0, position 0.

2 Answers

Answers 1

The error you are getting is a JSON.net one and happens during JSON deserialization, which means there is no problem with updating of the static variable, because the code never gets to that point (it ends on the catch (Newtonsoft.Json.JsonReaderException readerExp)).

This narrows your problem pretty well. There is very likely something wrong with the response you are receiving from the server. Put a breakpoint on the line var jsonReader = ... and check the contents of the page_result variable to see if they don't contain any unexpected characters. Potentially you can also dump the response into a JSON validator to confirm if it is actually valid (https://jsonlint.com/)

Answers 2

Each variable is scoped in a memory dedicated only for the thread where its declaration is performed in, so, when you're accessing that variable from another thread, you're reading a copy of that in the memory of your other thread.
This copy is performed when the thread is synchronized with the another, and not always this is done just when you set the variable.
Therefore, you have to add volatile modifier to your variable declaration, which signs that the variable must be allocated and deallocated in a global synchronization scope.
Try declaring your variable such as this:

public static volatile string username = ""; 

Read More

Destruction of condition variable randomly loses notification

Given a condition_variable as a member of a class, my understanding is that:

1. The condition variable is destroyed after the class destructor completes.
2. Destruction of a condition variable does not need to wait for notifications to have been received.

In light of these expectations, my question is: why does the example code below randomly fail to notify a waiting thread?

#include <mutex> #include <condition_variable> #define NOTIFY_IN_DESTRUCTOR   struct notify_on_delete {     std::condition_variable cv;      ~notify_on_delete() { #ifdef NOTIFY_IN_DESTRUCTOR         cv.notify_all(); #endif     } };  int main () {     for (int trial = 0; trial < 10000; ++trial) {         notify_on_delete* nod = new notify_on_delete();         std::mutex flag;         bool kill = false;          std::thread run([nod, &flag, &kill] () {             std::unique_lock<std::mutex> lock(flag);             kill = true;             nod->cv.wait(lock);         });          while(true) {             std::unique_lock<std::mutex> lock(flag);             if (!kill) continue; #ifdef NOTIFY_IN_DESTRUCTOR             delete nod; #else             nod->cv.notify_all(); #endif             break;         }         run.join(); #ifndef NOTIFY_IN_DESTRUCTOR         delete nod; #endif     }     return 0; } 

In the code above, if NOTIFY_IN_DESTRUCTOR is not defined then the test will run to completion reliably. However, when NOTIFY_IN_DESTRUCTOR is defined the test will randomly hang (usually after a few thousand trials).

I am compiling using Apple Clang: Apple LLVM version 9.0.0 (clang-900.0.39.2) Target: x86_64-apple-darwin17.3.0 Thread model: posix C++14 specified, compiled with DEBUG flags set.

EDIT:

To clarify: this question is about the semantics of the specified behavior of instances of condition_variable. The second point above appears to be reenforced in the following quote:

Blockquote Requires: There shall be no thread blocked on *this. [ Note: That is, all threads shall have been notified; they may subsequently block on the lock specified in the wait. This relaxes the usual rules, which would have required all wait calls to happen before destruction. Only the notification to unblock the wait needs to happen before destruction. The user should take care to ensure that no threads wait on *this once the destructor has been started, especially when the waiting threads are calling the wait functions in a loop or using the overloads of wait, wait_­for, or wait_­until that take a predicate. — end note ]

The core semantic question seems to be what "blocked on" means. My present interpretation of the quote above would be that after the line

cv.notify_all(); // defined NOTIFY_IN_DESTRUCTOR 

in ~notify_on_delete() the thread test is not "blocked on" nod - which is to say that I presently understand that after this call "the notification to unblock the wait" has occurred, so according to the quote the requirement has been met to proceed with the destruction of the condition_variable instance.

Can someone provide a clarification of "blocked on" or "notification to unblock" to the effect that in the code above, the call to notify_all() does not satisfy the requirements of ~condition_variable()?

2 Answers

Answers 1

When NOTIFY_IN_DESTRUCTOR is defined:
Calling notify_one()/notify_all() doesn't mean that the waiting thread is immediately woken up and the current thread will wait for the other thread. It just means that if the waiting thread wakes up at some point after the current thread has called notify, it should proceed. So in essence, you might be deleting the condition variable before the waiting thread wakes up (depending on how the threads are scheduled).

The explanation for why it hangs, even if the condition variable is deleted while the other thread is waiting on it lies on the fact the wait/notify operations are implemented using queues associated with the condition variables. These queues hold the threads waiting on the condition variables. Freeing the condition variable would mean getting rid of these thread queues.

Answers 2

I am pretty sure your vendors implementation is broken. Your program looks almost OK from the perspective of obeying the contract with the cv/mutex classes. I couldn’t 100% verify, I am behind one version.

Almost OK, because as Marek R points out, you are relying on referencing a class after its destruction has begun; not the cv/mutex class, your notify_on_delete class. The conflict is a bit academic. I doubt clang would depend upon nod remaining valid after control had transferred to nod->cv.wait(); but the real customer of most compiler vendors are benchmarks, not programmers.

As as general note, multi-threaded programming is difficult, and having now peaked at the c++ threading model, it might be best to give it a decade or two to settle down. It’s contracts are astonishing. When I first looked at your program, I thought ‘duh, there is no way you can destroy a cv that can be accessed because RAII’. Silly me.

Pthreads is another awful API for threading. At least it doesn’t attempt over-reach, and is mature enough that robust test suites keep vendors in line.

Read More

Safely “lend” memory block to another thread in C, assuming no “concurrent access”

The problem

I want to allocate memory in one thread, and safely "lend" the pointer to another thread so it can read that memory.

I'm using a high level language that translates to C. The high level language has threads (of unspecified threading API, since it's cross-platform -- see below) and supports standard C multi-threading primitives, like atomic-compare-exchange, but it's not really documented (no usage examples). The constraints of this high-level language are:

• Each thread executes an event-processing infinite loop.
• Each thread has it's own local heap, managed by some custom allocator.
• Each thread has one "input" message queue, that can contain messages from any number of different other threads.
• The message passing queues are:
  1. For fixed-type messages
  2. Using copying

Now this is impractical for large (don't want the copy) or variable-sized (I think array-size is part of the type) messages. I want to send such messages, and here's the outline of how I want to achieve it:

• A message (either a request or a reply) can either store the "payload" inline (copied, fixed limit on total values size), or a pointer to data in the sender's heap
• The message contents (data in sender's heap) is owned by the sending thread (allocate and free)
• The receiving thread sends an ack to the sending thread when they are done with the message content
• The "sending" threads must not modify the message contents after sending them, until receiving the (ack).
• There should never be a concurrent read access on memory being written to, before the writing is done. This should be guaranteed by the message queues work-flow.

I need to know how to ensure that this works without data races. My understanding is that I need to use memory fences, but I'm not entirely sure which one (ATOMIC_RELEASE, ...) and where in the loop (or if I need any at all).

---

Portability considerations

Because my high-level language needs to be cross-platform, I need the answer to work on:

• Linux, MacOS, and optionally Android and iOS
  • using pthreads primitives to lock message queues: pthread_mutex_init and pthread_mutex_lock + pthread_mutex_unlock
• Windows
  • using Critical Section Objects to lock message queues: InitializeCriticalSection, and EnterCriticalSection + LeaveCriticalSection

If it helps, I'm assuming the following architectures:

• Intel/AMD PC architecture for Windows/Linux/MacOS(?).
• unknown (ARM?) for iOS and Android

And using the following compilers (you can assume a "recent" version of all of them):

• MSVC on Windows
• clang on Linux
• Xcode On MacOS/iOS
• CodeWorks for Android on Android

I've only built on Windows so far, but when the app is done, I want to port it to the other platforms with minimal work. Therefore I'm trying to ensure cross-platform compatibility from the start.

---

Attempted Solution

Here is my assumed work-flow:

1. Read all the messages from the queue, until it's empty (only block if it was totally empty).
2. Call some "memory fence" here?
3. Read the messages contents (target of pointers in messages), and process the messages.
   • If the message is a "request", it can be processed, and new messages buffered as "replies".
   • If the message is a "reply", the message content of the original "request" can be freed (implicit request "ack").
   • If the message is a "reply", and it itself contains a pointer to "reply content" (instead of an "inline reply"), then a "reply-ack" must be sent too.
4. Call some "memory fence" here?
5. Send all the buffered messages into the appropriate message queues.

Real code is too large to post. Here is simplified (just enough to show how the shared memory is accessed) pseudocode using a mutex (like the message queues):

static pointer p = null static mutex m = ... static thread_A_buffer = malloc(...)  Thread-A:   do:     // Send pointer to data     int index = findFreeIndex(thread_A_buffer)     // Assume different value (not 42) every time     thread_A_buffer[index] = 42     // Call some "memory fence" here (after writing, before sending)?     lock(m)     p = &(thread_A_buffer[index])     signal()     unlock(m)     // wait for processing     // in reality, would wait for a second signal...     pointer p_a = null     do:       // sleep       lock(m)       p_a = p       unlock(m)     while (p_a != null)     // Free data     thread_A_buffer[index] = 0     freeIndex(thread_A_buffer, index)   while true  Thread-B:   while true:     // wait for data     pointer p_b = null     while (p_b == null)       lock(m)       wait()       p_b = p       unlock(m)     // Call some "memory fence" here (after receiving, before reading)?     // process data     print *p_b     // say we are done     lock(m)     p = null     // in reality, would send a second signal...     unlock(m) 

Would this solution work? Reformulating the question, does Thread-B print "42"? Always, on all considered platforms and OS (pthreads and Windows CS)? Or do I need to add other threading primitives such as memory fences?

---

Research

I've spent hours looking at many related SO questions, and read some articles, but I'm still not totally sure. Based on @Art comment, I probably don't need to do anything. I believe this is based on this statement from the POSIX standard, 4.12 Memory Synchronization:

[...] using functions that synchronize thread execution and also synchronize memory with respect to other threads. The following functions synchronize memory with respect to other threads.

My problem is that this sentence doesn't clearly specify if they mean "all the accessed memory", or "only the memory accessed between lock and unlock." I have read people arguing for both cases, and even some implying it was written imprecisely on purpose, to give compiler implementers more leeway in their implementation!

Furthermore, this applies to pthreads, but I need to know how it applies to Windows threading as well.

I'll choose any answer that, based on quotes/links from either a standard documentation, or some other highly reliable source, either proves that I don't need fences or shows which fences I need, under the aforementioned platform configurations, at least for the Windows/Linux/MacOS case. If the Windows threads behave like the pthreads in this case, I'd like a link/quote for that too.

The following are some (of the best) related questions/links I read, but the presence of conflicting information causes me to doubt my understanding.

• Does pthread_mutex_lock contains memory fence instruction?
• Memory Fences - Need help to understand
• Problem with pThread sync issue
• Memory Visibility Through pthread Library?
• clarifications on full memory barriers involved by pthread mutexes
• Memory model spec in pthreads
• http://www.hpl.hp.com/techreports/2005/HPL-2005-217R1.html
• http://pubs.opengroup.org/onlinepubs/9699919799/basedefs/V1_chap04.html#tag_04_11
• https://msdn.microsoft.com/en-us/library/windows/desktop/ms684208(v=vs.85).aspx

0 Answers

Read More

How to use the threading module to call a function?

Or in other words, how to create a time delayed function?
I have a python bot that is supposed to send notifications to user's followers upon the usage of certain commands.

For example , if Tim runs the command ' >follow Tom ', all of Tim's followers will be notified in PM's that he followed Tom , and Tom will be notified that Tim followed him.

I have tested this function with a large amount of followers , and the bot remains stable and avoids being kicked from the server , i'm guessing because the for loop adds a delay to each message sent to each follower.

The problem I have is if two user's were to simultaneously run a command that warrant's a notification. The bot gets kicked offline immediately. So what I need is to add an artificial delay before the notification function is run. Time.sleep() , does not work. all it does it freeze the entire program , and hold every command in a queue. (If two user's ran >follow , it would sleep for 2 seconds , and just run both their commands after the delay)

I'm trying to use the threading module in order to replace time.sleep(). My notification function is the following.

#message is the message to be sent, var is the username to use def notify(message,var):       #SQL connect        dbconfig = read_db_config()       conn = MySQLConnection(**dbconfig)       cursor = conn.cursor()       #choose all of user's followers       cursor.execute('select username from users where notifications=0 and username IN (select follower from followers where followed like "{}")'.format(var))       results = cursor.fetchall()       #for each , send a PM       for result in results:         self.pm.message(ch.User(str(result[0])), message)       conn.close()   

So how would I use threading to do this? I've tried a couple ways , but let's just go with the worst one.

def example(_):     username = 'bob'     # _ is equal to args     a = notify("{} is now following {}.".format(username,_),username)     c =threading.Timer(2,a)     c.start() 

This will throw a Nonetype error in response.

Exception in thread Thread-1: Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner self.run() File "/usr/lib/python2.7/threading.py", line 1082, in run self.function(*self.args, **self.kwargs) TypeError: 'NoneType' object is not callable

Note: I think this method will work , there will be a lot of users using the bot at once, so until it breaks this seems like a fix.

2 Answers

Answers 1

Here is some code that may help. Notice the change to how notify handles the results.

import threading import Queue  def notifier(nq):     # Read from queue until None is put on queue.     while True:         t = nq.get()         try:             if t is None:                 break             func, args = t             func(*args)             time.sleep(2) # wait 2 seconds before sending another notice         finally:             nq.task_done()   # message is the message to be sent, var is the username to use, nq is the # queue to put notification on. def notify(message, var, nq):     #SQL connect     dbconfig = read_db_config()     conn = MySQLConnection(**dbconfig)     cursor = conn.cursor()     #choose all of user's followers     cursor.execute('select username from users where notifications=0 and username IN (select follower from followers where followed like "{}")'.format(var))     results = cursor.fetchall()     #for each , send a PM     for result in results:         # Put the function to call and its args on the queue.         args = (ch.User(str(result[0])), message)         nq.put((self.pm.message, args))     conn.close()   if __name__ == '__main__':     # Start a thread to read from the queue.     nq = Queue.Queue()     th = threading.Thread(target=notifier, args=(nq,))     th.daemon = True     th.start()     # Run bot code     # ...     #     nq.put(None)     nq.join() # block until all tasks are done 

Answers 2

I would try using a threading lock like the class I wrote below.

This will cause only one thread to be able to send PM's at any given time.

class NotifyUsers():     def __init__(self, *args, **kwargs):         self.notify_lock = threading.Lock()         self.dbconfig = read_db_config()         self.conn = MySQLConnection(**dbconfig)         self.cursor = self.conn.cursor()      def notify_lock_wrapper(self, message, var):         self.notify_lock.acquire()         try:             self._notify(message, var)         except:             # Error handling here             pass         finally:             self.notify_lock.release()      def _notify(self, message, var):         #choose all of user's followers         self.cursor.execute('select username from users where notifications=0 and username IN (select follower from followers where followed like "{}")'.format(var))         results = self.cursor.fetchall()          #for each, send a PM         for result in results:             self.pm.message(ch.User(str(result[0])), message) 

Read More

Is code between ConcurrentQueue wait_and_pop and push thread safe?

ConcurrentQueue.wait_and_pop(detector); detector->detect(); ConcurrentQueue.push(detector); 

detector->detect() is not thread safe.

1. Is the code safe in multithread env if ConcurrentQueue is implemented by mutex? Why?

2. Is the code safe in multithread env if ConcurrentQueue is implemented by lock free? Why?

0 Answers

Read More