Moving forward with pthreads.

Author: Anton Yarkov

multithreading/01 - Fork Processes And End Differently.c

@@ -6,7 +6,7 @@
 #include "Common.h"
 
 // Compilation:
-// gcc -std=gnu99 "01 - Fork Processes And End Differently.c" -o forkEndProcesses
+// gcc -std=gnu99 ErrorHandling.c LogF.c "01 - Fork Processes And End Differently.c" -o forkEndProcesses
 
 // Task:
 // Use fork to create a child processes, which stop working with different reasons.

multithreading/04 - Process attributes.c

@@ -6,11 +6,98 @@
 #include "Common.h"
 
 // Compilation:
-// gcc -std=gnu99 "04 - Process attributes.c" -o processAttrs
+// gcc -std=gnu99 ErrorHandling.c LogF.c "04 - Process attributes.c" -o processAttrs
 
 // Task:
 // Work with process attributes.
 
+static void printLimitValue(rlim_t limit)
+{
+    if (limit == RLIM_INFINITY)
+        printf("RLIM_INFINITY");
+    else
+        printf("%llu", (unsigned long long)limit);
+}
+
+static bool readLimits(int resource, const char *name)
+{
+    struct rlimit r;
+    ec_neg1( getrlimit(resource, &r) )
+    printf("%s: ", name);
+    printf(" current: ");
+    printLimitValue(r.rlim_cur);
+    printf(" max: ");
+    printLimitValue(r.rlim_max);
+    printf("\n");
+    
+    return true;
+    
+EC_CLEANUP_BGN
+    return false;
+EC_CLEANUP_END
+}
+
+static bool readAndChangeProcessLimits()
+{
+    printf("6 CHILD: Changing process limits...\n");
+    struct rlimit r;
+    int fd;
+    char buf[500] = { 0 };
+    if (sizeof(rlim_t) > sizeof(long long))
+        printf("Warning: rlim_t > long long, results might be wrong!\n");
+    
+    ec_false( readLimits(RLIMIT_CORE, "6 CHILD: RLIMIT_CORE") )
+    ec_false( readLimits(RLIMIT_CPU, "6 CHILD: RLIMIT_CPU") )
+    ec_false( readLimits(RLIMIT_DATA, "6 CHILD: RLIMIT_DATA") )
+    ec_false( readLimits(RLIMIT_FSIZE, "6 CHILD: RLIMIT_FSIZE") )
+    ec_false( readLimits(RLIMIT_NOFILE, "6 CHILD: RLIMIT_NOFILE") )
+    ec_false( readLimits(RLIMIT_STACK, "6 CHILD: RLIMIT_STACK") )
+    
+    ec_neg1( getrlimit(RLIMIT_FSIZE, &r) )
+    
+    r.rlim_cur = 500;
+    
+    ec_neg1( setrlimit(RLIMIT_FSIZE, &r) )
+    ec_neg1( fd = open("tmp", O_WRONLY | O_CREAT | O_TRUNC, PERM_FILE) )
+    ec_neg1( write(fd, buf, sizeof(buf)) )
+    ec_neg1( write(fd, buf, sizeof(buf)) )
+    printf("6 CHILD: 2 buffers has been written!\n");
+    
+    return true;
+    
+EC_CLEANUP_BGN
+    return false;
+EC_CLEANUP_END
+}
+
+static bool showProcessResourcesStat()
+{
+    struct rusage r;
+    ec_neg1( getrusage(RUSAGE_SELF, &r) )
+    printf("7 CHILD: user CPU time used %d \n", r.ru_utime);
+    printf("7 CHILD: system CPU time used %d \n", r.ru_stime);
+    printf("7 CHILD: max resident set size %d \n", r.ru_maxrss);
+    printf("7 CHILD: integral shared size %d \n", r.ru_ixrss);
+    printf("7 CHILD: integral unshared data (data segment) size %d \n", r.ru_idrss);
+    printf("7 CHILD: integral unshared stack size %d \n", r.ru_isrss);
+    printf("7 CHILD: page reclaims (soft page faults) %d \n", r.ru_minflt);
+    printf("7 CHILD: hard page faults %d \n", r.ru_majflt);
+    printf("7 CHILD: swaps / page file touches %d \n", r.ru_nswap);
+    printf("7 CHILD: block input operations %d \n", r.ru_inblock);
+    printf("7 CHILD: block output operations %d \n", r.ru_oublock);
+    printf("7 CHILD: IPC messages sent %d \n", r.ru_msgsnd);
+    printf("7 CHILD: IPC messages received %d \n", r.ru_msgrcv);
+    printf("7 CHILD: signals received %d \n", r.ru_nsignals);
+    printf("7 CHILD: voluntary context switches by process %d \n", r.ru_nvcsw);
+    printf("7 CHILD: involuntary context switches by core %d \n", r.ru_nivcsw);
+    
+    return true;
+    
+EC_CLEANUP_BGN
+    return false;
+EC_CLEANUP_END
+}
+
 int main(int argc, char *argv[])
 {
     printf("Enter to main!\n");
@@ -39,10 +126,14 @@ int main(int argc, char *argv[])
             printf("5 CHILD: Let's decrease Priority Level for N=5: %d (N-20)\n", nice(5));
             printf("5 CHILD: Let's increase Priority Level for N=-5: %d (N-20)\n", nice(-5));
 
-            printf("6 CHILD: Wait 4 seconds...\n");
+            readAndChangeProcessLimits();
+            
+            showProcessResourcesStat();
+            
+            printf("8 CHILD: Wait 4 seconds...\n");
             sleep(4);
-            printf("7 CHILD: Exit!\n");
-            exit(EXIT_FAILURE);
+            printf("9 CHILD: Exit!\n");
+            exit(EXIT_SUCCESS);
         default:
             printf("1 PARENT: Parent process %d!\n", getpid());
             printf("2 PARENT: Child PID %d\n", pid);
@@ -57,4 +148,8 @@ int main(int argc, char *argv[])
 
     printf("Exit process!\n");
     exit(EXIT_SUCCESS);
+    
+EC_CLEANUP_BGN
+    exit(EXIT_FAILURE);
+EC_CLEANUP_END
 }

multithreading/05 - PThreads.c

@@ -6,15 +6,15 @@
 // gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "05 - PThreads.c" -o pthreads
 
 // Task:
-// Simple example of a counter working in 2 threads.
+// Simple example of a counter working in 2 threads. Example shows that sequence of ++ calls is not deterministic.
 
 static long x = 0;
 
 static void* thread_func(void* arg)
 {
     while (true)
     {
-        printf("Thread 2, counter value %ld\n", ++x);
+        printf("Thread 2, counter value %ld\n", ++x); // !!! WARNING!!! 2 Threads use the same variable. But increment operation is not atomic.
         sleep(1);
     }
 }
@@ -23,11 +23,15 @@ int main(void)
 {
     pthread_t tid;
 
-    ec_rv(pthread_create(&tid, NULL, thread_func, NULL))
+    // Use macros ec_rv to determine returned error (if happened), because pthread doesn't use errno variable to set error info.
+    ec_rv(pthread_create(&tid,
+        NULL, // Use attributes by default. To create attribute, call pthread_attr_setscope or pthread_attr_setstaoksize
+        thread_func,
+        NULL))
 
     while (x < 10)
     {
-        printf("Thread 1, counter value %ld\n", ++x);
+        printf("Thread 1, counter value %ld\n", ++x); // !!! WARNING!!! 2 Threads use the same variable. But increment operation is not atomic.
         sleep(2);
     }
     return EXIT_SUCCESS;

multithreading/06 - PThreads Joinable BAD.c renamed to multithreading/06 - PThreads Joinable.c

@@ -6,11 +6,9 @@
 // gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "06 - PThreads Joinable BAD.c" -o joinable_threads
 
 // Task:
-// Simple example of an application working with 2 threads, where parent waits results from child.
+// Simple example of an application working with 2 threads, where parent WAITS for a results from child.
 // Thread 2 stops calculation if common value already more than limit.
 
-// WARNING! Code contains a problem marked with comment.
-
 static long x = 0;
 
 static void* thread_func(void* arg)

multithreading/07 - PThreads Joinable with Sync.c renamed to multithreading/07 - PThreads Joinable Mutexed.c

@@ -3,15 +3,17 @@
 #include "Defs.h"
 
 // Compilation:
-// gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "07 - PThreads Joinable with Sync.c" -o joinable_threads_synced
+// gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "07 - PThreads Joinable Mutexed.c" -o joinable_threads_synced
 
 // Task:
-// Simple example of an application working with 2 threads, where parent waits results from child.
+// Simple example of an application working with 2 threads, where parent WAITS for a results from child.
 // Thread 2 stops calculation if common value already more than limit.
 
 // Access to variable is controlled with mutex.
 
+// Can be static, extern and even auto. In case of auto need to initialize with function pthread_mutex_init, not by a constant (for thread safety).
 static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
+
 static long x = 0;
 
 static void* thread_func(void* arg)
@@ -34,6 +36,8 @@ static void* thread_func(void* arg)
     return (void *)x;
 
 EC_CLEANUP_BGN
+    // To avoid dead lock: Thread 2 faulted on mutex_unlock, this Thread 2 is locked forever.
+    (void)pthread_mutex_unlock(&mtx);
     EC_FLUSH("thread_func")
     return NULL;
 EC_CLEANUP_END

multithreading/08 - PThreads Joinable with Sync improved.c renamed to multithreading/08 - PThreads Joinable Mutexed improved.c

@@ -3,7 +3,7 @@
 #include "Defs.h"
 
 // Compilation:
-// gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "08 - PThreads Joinable with Sync improved.c" -o joinable_threads_synced1
+// gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "08 - PThreads Joinable Mutexed improved.c" -o joinable_threads_synced1
 
 // Task:
 // Simple example of an application working with 2 threads, where parent waits results from child.
@@ -17,7 +17,7 @@ static long get_and_incr_x(long incr)
                                                             // And C++ may even use lazy initialization for mutex.
     static long x = 0;
 
-    // rtn and incr variables are on stack - i.e. local
+    // rtn and incr variables are on stack - i.e. local, so we need to use mutex for access to it.
     long rtn;
 
     ec_rv( pthread_mutex_lock(&mtx) )
@@ -36,7 +36,7 @@ static void* thread_func(void* arg)
     long val = 0;
     while (val < (long)arg)
     {
-        val = get_and_incr_x(1);
+        val = get_and_incr_x(1); // Use local variable, because it is thread safe.
         printf("Thread 2, counter value %ld\n", val);
         sleep(1);
     }
@@ -56,7 +56,7 @@ int main(void)
     long val = 0;
     while (val < 10)
     {
-        val = get_and_incr_x(1);
+        val = get_and_incr_x(1); // Use local variable, because it is thread safe.
         printf("Thread 1, counter value %ld\n", val);
         sleep(2);
     }

multithreading/09 - PThreads Mutexed with Conditional Variables.c

@@ -0,0 +1,100 @@
+#include <pthread.h>
+
+#include "Defs.h"
+
+// Compilation:
+// gcc -std=gnu99 -pthread ErrorHandling.c LogF.c "09 - PThreads Mutexed with Conditional Variables.c" -o threads_and_queue
+
+// Task:
+// Application working with 2 threads - one is putting data into queue, another one gets data from the queue.
+// a. Threads to be stopped forcibly with pthread_cancel.
+// b. Queue data do not contain mark of END of data flow for normal cases (we use (a) for both
+// normal and abnormal termination). This is TO BE DONE.
+// c. We may use Signals for normal and abnormal termination. But it has a problems:
+
+// Access to queue is controlled with mutex and conditional variable.
+
+static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
+
+struct node
+{
+    int n_number;
+    struct node *n_next;
+} *head = NULL;
+
+static void cleanup_handler(void *arg)
+{
+    free(arg);
+    (void)pthread_mutex_unlock(&mtx);
+}
+
+static void* thread_func(void* arg)
+{
+    struct node *p = NULL; // Should be initialized at least with NULL, since we use free.
+    
+    pthread_cleanup_push(cleanup_handler, p); // Set up cleanup handler for case of forced thread termination
+                                              // to avoid dead lock on locked mutex.
+    while (true)
+    {
+        ec_rv( pthread_mutex_lock(&mtx) )
+        
+        while (head == NULL) // Loop is used because pthread_cond_wait
+                             // can be interrupted by signal (SIGINT, etc.)
+                             // and we need to get back to conditional wait.        
+            ec_rv( pthread_cond_wait(&cond, &mtx) ) // Conditional variable UNLOCKS mutex while waiting,
+                                                    // but locks it back when condition happened.
+        
+        p = head;
+        head = head->n_next;
+        int number = p->n_number;
+        free(p); // Free called here, because next point - printf(), might be a
+                 // Point of Interruption (free() or malloc() cannot),
+                 // and we need to make sure memory is freed.
+        printf("Got number %d from the queue\n", number);
+        ec_rv( pthread_mutex_unlock(&mtx) )
+    }
+    
+    pthread_cleanup_pop(cleanup_handler, p); 
+    return (void *)true;
+    
+EC_CLEANUP_BGN
+    (void)pthread_mutex_unlock(&mtx);
+    EC_FLUSH("thread_func")
+    return (void *)false;
+EC_CLEANUP_END
+}
+
+int main(void)
+{
+    pthread_t tid;
+    int i;
+    struct node *p;
+    
+    ec_rv( pthread_create(&tid, NULL, thread_func, NULL) )
+
+    for (i = 0; i < 10; i++)
+    {
+        // Imitation of data receive here...
+        ec_null( p = malloc(sizeof(struct node)) )
+        p->n_number = i;
+        
+        // Put into queue.... Using mutex protection.
+        ec_rv( pthread_mutex_lock(&mtx) )
+        p->n_next = head;
+        head = p;
+        ec_rv( pthread_cond_signal(&cond) ) // Signal about data in the queue.
+        ec_rv( pthread_mutex_unlock(&mtx) )
+        sleep(1);
+    }
+    
+    ec_rv( pthread_cancel(tid) )    
+    ec_rv( pthread_join(tid, NULL) )
+    printf("End of transmission\n");
+    
+    return EXIT_SUCCESS;
+
+EC_CLEANUP_BGN
+    return EXIT_FAILURE;
+EC_CLEANUP_END
+}